Data Warehousing and Analytics Platform

7/1/2026

This post details the evolution of Meta's BLOB-storage architecture to specifically address the demands of AI workloads. It highlights the challenges of storage bottlenecks impacting GPU utilization and research velocity, and introduces a rebuilt foundation with a unified metadata schema, a fat client SDK eliminating dataplane proxy, and regional deployments. It also covers strategies for handling spikes and hot spots with distributed data caches and readplan metadata caches, and protocol optimizations like hedged reads and dynamic concurrency control.

5/12/2026

This post details the successful migration of Meta's large-scale data ingestion system, which processes petabytes of social graph data daily. It outlines the challenges of migrating thousands of jobs, the phased migration lifecycle (shadow phase, reverse shadow phase, cleanup), custom data quality analysis tooling using row count and checksum comparisons, and strategies for handling rollout and rollback with change data capture (CDC). The post also describes automated tooling for monitoring and migrating jobs in batches, and planning for limited capacity by categorizing jobs based on throughput and priority.

10/6/2025

This post introduces OpenZL, an open-source format-aware compression framework designed for structured data. It aims to provide the performance of specialized compressors with the maintenance simplicity of a single binary. OpenZL achieves this by applying a configurable sequence of transforms to reveal patterns in structured data before compression, offering significant improvements in compression ratio and speed compared to generic compressors like Zstandard and xz, especially for structured datasets. It also introduces a trainer component for automatically generating optimal compression configurations and a universal decoder that remains consistent even as compression plans evolve.

8/14/2025

This post introduces the integration of AI agents into Meta's data warehouse to enhance data access productivity and security. It details the development of user and owner agents, composed of specialized sub-agents, to streamline data discovery, access requests, and permission management. The post also explains how the data warehouse is adapted for agentic systems, including context and intention management, and introduces a 'partial data preview' use case enabled by these agents, emphasizing query-level access control, data-access budgets, and rule-based risk management.

5/8/2025

This post details the integration of NVIDIA cuVS into Meta's open-source Faiss library (v1.10), significantly accelerating GPU-based vector search. It highlights performance improvements in both inverted file (IVF) and graph indexing, showcasing substantial speedups in build times and reductions in search latency through collaboration with NVIDIA. This advancement extends Meta's capabilities in large-scale similarity search and AI infrastructure.

4/28/2025

This post details Meta's advancements in data understanding as part of its Privacy Aware Infrastructure (PAI). Key contributions include the development of a "shift-left" approach integrating data schematization and annotations early in product development, the creation of a universal privacy taxonomy for standardized data privacy management, and the establishment of OneCatalog for discovering, registering, and enumerating data assets. The post also outlines solutions to challenges such as understanding data at scale through a shared asset schema format, ensuring consistent definitions with a unified taxonomy of semantic types, improving annotation quality by combining schematization with code annotations and multiple classification signals, and overcoming organizational barriers through collaboration and intuitive tooling. A walkthrough of understanding user data for the "Beliefs" feature in Facebook Dating illustrates the five-step approach: schematizing data using DataSchema (based on Thrift IDL), predicting metadata at scale through a universal privacy taxonomy and data classification, and applying these principles across distributed systems and data warehouses.

2/4/2025

Introduced data logs as a new capability within the Download Your Information (DYI) tool, enabling users to access granular data previously only available in the data warehouse (Hive). This involved developing a system to batch user requests, perform optimized INNER JOINs between user ID tables and data partitions in Hive, and post-process raw data into user-friendly HTML files. Key technical challenges addressed include the inefficiency of individual queries on massive Hive tables, the computational and storage costs of caching, and ensuring data correctness through verification steps to mitigate Spark concurrency bugs.

1/23/2025

This post details Meta's development of a scalable data lineage system to support its Privacy Aware Infrastructure (PAI) initiative. It describes techniques for collecting data flow signals, including static code analysis and runtime instrumentation via 'Privacy Probes,' which capture and compare source and sink payloads. The post illustrates the implementation for tracking 'religion' data in the Facebook Dating app, covering its propagation through web systems and into the data warehouse for offline processing, highlighting the use of Policy Zones for privacy controls.

9/17/2024

This post introduces Bento, Meta's internal distribution of Jupyter Notebooks, highlighting its use for prototyping, machine learning workflows, and enabling advanced features like scheduled notebooks and serverless execution via WebAssembly. This expands the data infrastructure narrative by detailing a key tool for interactive data analysis and development.

6/10/2024

Introduced serverless Jupyter notebooks for lite workloads within the Bento platform by integrating Pyodide and WebAssembly. This enables in-browser code execution, removing the need for remote compute provisioning for prototyping and simple ML workflows. Implemented a kernel bridge abstraction for seamless switching between server-based and serverless kernels, and adapted cell magics (e.g., %%sql, %%googlesheet, %%graphql) to run in the browser context. Utilized web workers to prevent browser lock-up during kernel operations and integrated with Tupperware for data uploads.

5/22/2024

This post details Meta's shift towards a composable data management architecture to address fragmentation and improve engineering efficiency. It introduces the concept of decomposing data systems into reusable components like language frontends, intermediate representations, query optimizers, execution engines, and runtimes. The post highlights the open-source Velox execution engine as a key enabler, demonstrating significant performance improvements (3-10x) in integrations with systems like Presto and Spark. Velox leverages columnar and vectorized execution, compressed execution, lazy materialization, and adaptivity. The post also emphasizes the importance of open standards like Apache Arrow for interoperability and discusses the ongoing work in system-wide integration.

3/18/2024

Introduced Logarithm, a new internal, serverless, multitenant logging engine designed for AI training workflows and services. Logarithm ingests and indexes over 100GB/s of logs in real-time, supporting thousands of queries per second with service-level guarantees on freshness, completeness, durability, and query latency. It supports unstructured text logs with optional metadata key-value pairs, including typed structures and regex-based extraction. The system architecture separates compute (ingestion and query) from storage, utilizes tiered storage (memory, local SSD, blob storage), and employs lightweight, disaggregated secondary indices with Bloom filters. Logarithm's design prioritizes scalability, fault tolerance, and efficient handling of high-throughput, time-ordered log data, with a focus on enabling debugging of complex AI training failures by collocating system and model telemetry.

2/20/2024

This post details the convergence of Meta's open-source execution engine, Velox, with Apache Arrow. It introduces three new Arrow format layouts (StringView, ListView, and Run-End-Encoding) inspired by Velox's internal representations. These additions aim to improve interoperability and efficiency in data management systems by enabling zero-copy in-memory data transfer and supporting out-of-order writes for variable-sized data types, contributing to a more composable data management architecture.

10/31/2023

This post introduces the subsystem of Meta's Systematic Code and Asset Removal Framework (SCARF) responsible for automating the identification and safe removal of unused data types. It details how SCARF scans production data systems to identify unused tables or assets, measuring both static code references (using Glean) and runtime access patterns to determine safety. The process involves notifying engineering teams, a blocking period for reads/writes as a dry run, and eventual deletion. The post also addresses challenges in handling cross-system dependencies and code usage, and highlights SCARF's scale in removing petabytes of data and millions of data types daily.

8/29/2023

Introduced a scheduled notebooks infrastructure by integrating the internal Bento Jupyter notebooks platform with the Dataswarm batch ETL pipeline framework. This system enables automated execution of notebooks for ETL, data transformations, and visualizations. Key technical contributions include the development of a custom BentoOperator that runs notebook executions in network-isolated containers, uses custom cells for structured data fetching (e.g., SQL cells) and writing (e.g., pandas.DataFrame.to_csv()), and integrates with Meta's data purpose framework for privacy compliance. The system also includes UI for schedule setup, a diff review process, and a debugging interface.

5/16/2023

This post details the implementation of MySQL Raft at Meta, a system that integrates the Raft consensus engine with MySQL to create a replicated state machine. It replaces the previous semisynchronous replication protocol with a Raft-based approach, enhancing reliability, safety, failover time, and operational simplicity. Key technical contributions include the development of the kuduraft library (a fork of Apache Kudu with enhancements like FlexiRaft, proxying, compression, log abstraction, and primary ban), the MyRaft MySQL plugin for interfacing with Raft, and novel replication topologies and write transaction flows leveraging Raft for consensus. The post also describes modifications to MySQL's binary log and crash recovery mechanisms to support the Raft protocol.

3/9/2023

Introduced Velox, an open-source unified execution engine designed to accelerate data management systems and streamline their development. Velox unifies common data-intensive components of data computation engines, providing a framework for consistent semantics and reducing work duplication. It integrates with existing Meta systems like Presto, Spark, and PyTorch (via TorchArrow), and has attracted over 150 contributors from the open-source community. Velox's architecture includes a generic type system, Apache Arrow-compatible columnar memory layout, vectorized expression evaluation, extensible function and operator APIs, I/O interfaces for connectors and file formats, and resource management primitives. Experimental results show significant speedups in data querying and CPU-bound queries when integrated with Presto (Prestissimo) and Spark (Gluten).

1/26/2023

This post details the engineering challenges and solutions for migrating Meta's exabyte-scale data platform to the Tulip serialization format. It describes the motivation for replacing legacy formats (Hive Text Delimited, JSON) due to reliability and efficiency issues, and outlines the migration strategy. Key technical contributions include the development of a mixed-mode wire format to support gradual migration, data consumption readers and format converters to abstract serialization from application code, a CLI debugging tool (loggertail) for post-migration validation, shadow loggers for end-to-end acceptance testing, and mechanisms for rate limiting and partial rollouts to manage risk.

11/30/2022

This post introduces UPM (Unified Programming Model), an internal library for static analysis of SQL code. UPM represents SQL queries as semantic trees, enabling the development of SQL linters, data lineage analysis, and a unified SQL language front end across Meta's various query engines (Presto, Spark, XStream). It also enhances type-checking by leveraging user-defined types in table schemas to catch semantic errors like unit and ID comparison mismatches. The data lineage team uses UPM to build column-level data lineage graphs across the warehouse.

11/9/2022

Introduced Tulip, a new binary serialization protocol designed for Meta's data platform. Tulip addresses challenges in schema evolution, reliability, and efficiency by replacing legacy formats like Hive Text Delimited and JSON. It utilizes Thrift's TCompactProtocol and assigns field IDs through a data platform management module to ensure forward and backward compatibility, reduce data size (40-85% reduction), and decrease CPU cycles for serialization/deserialization (50-90% reduction).

9/7/2022

Open-sourced TAOBench, a benchmark framework and workload configurations designed to accurately model social network graph workloads. TAOBench captures transactional requirements, data colocation preferences, request distributions, and multitenant behavior by profiling requests from Meta's TAO graph data store. The benchmark demonstrates its utility by reporting results for five popular distributed databases and highlighting its use cases for testing new features, optimizations, and speculative workloads.

6/8/2022

Introduced Zelos, a ZooKeeper API built on the Delos distributed system platform. Zelos aims to replace Meta's existing ZooKeeper clusters by providing a more modular, extensible, and performant variant. Key technical contributions include addressing the impedance mismatch between ZooKeeper's session management and Delos's linearizable log by implementing speculative execution for session ordering and a RequestProcessor for ordering reads within sessions. It also details the challenges and solutions for transparently migrating ZooKeeper-based applications to the new Zelos architecture.

5/4/2022

Introduced Delta, a new object storage service designed for critical build and distribution artifacts, prioritizing high availability and strong consistency. Delta leverages chain replication for data redundancy and fault tolerance, trading latency and storage efficiency for simplicity and reliability. It supports basic operations (put, get, delete, list) and is horizontally scalable. The post details the architecture of Delta, its comparison with quorum replication, and its failure and recovery modes, including host failure detection and recovery mechanisms.

4/26/2022

Introduced SQL Notebooks, a new tool that merges SQL IDEs and Jupyter Notebooks for scalable and secure data analytics. This system allows for modular SQL queries using named cells that reference each other, supports sandboxed Python for post-processing and visualization, and enforces security and data freshness through controlled snapshotting and scheduled updates. It aims to deprecate the older Daiquery tool and integrate with the Bento/Jupyter notebook ecosystem.

3/7/2022

This post details improvements to LogDevice, Meta's scalable distributed log storage system, by augmenting its Flexible Paxos consensus algorithm to enhance read availability. It addresses the trade-offs in Flexible Paxos concerning read and write availability, particularly in the context of failures and maintenance. The post explains how LogDevice's recovery process and a separate membership mechanism using ZooKeeper were adapted to maintain write availability while improving read availability, even in scenarios where a majority of acceptors are unavailable, by allowing reads without a strict quorum and addressing 'stuck recovery' issues.

8/18/2021

Introduced RAMP-TAO, a new protocol that layers atomic transaction guarantees on top of TAO, Facebook's online social graph store. RAMP-TAO addresses the challenge of providing transactional semantics over an eventually consistent, massive-scale data store by focusing on guarding against fractured reads for recently updated data, minimizing performance and storage overheads for existing non-transactional workloads. It achieves over 99.9 percent of reads completing in one round-trip to the local cache with tail latency on par with existing TAO reads, incurring overhead only for opt-in applications.

8/6/2021

This post introduces ZippyDB, a large-scale, strongly consistent, geographically distributed key-value store at Facebook, built on RocksDB. It details ZippyDB's history, architecture (tiers, regions, shards, μshards, Akkio), data model (key-value, TTL, transactions, conditional writes), and configurable consistency/durability levels. ZippyDB addresses the need for a unified, scalable key-value store by reusing existing infrastructure like Data Shuttle, Shard Manager, and ZooKeeper, thereby reducing duplicated efforts across teams and enabling faster product development.

7/22/2021

This post details the technical challenges and solutions involved in migrating Facebook's core workloads from MySQL 5.6 to MySQL 8.0. It covers porting over 1,700 custom code patches, ensuring replication compatibility, minimizing application query changes, and fixing performance regressions. The migration involved a phased approach for replica sets, standardization on row-based replication, and extensive automation validation for both infrastructure and application queries. Specific issues encountered included binlog event format incompatibilities, conflicting error codes, reliance on .frm files, and dynamic privilege support. Application-level issues like reserved keyword conflicts and REGEXP incompatibilities were also addressed.

7/9/2021

Introduced the Ribbon filter, a new space-efficient data structure that is practically smaller than Bloom filters. Ribbon filters offer near-continuous configurability in the accuracy-versus-space trade-off, with space overhead as low as 1 percent or less, and O(1) query times. They save roughly 1/3 of memory compared with Bloom filters by solving a linear system using Gaussian elimination on a band matrix structure, with an on-the-fly construction algorithm called Rapid Incremental Boolean Banding ON the fly (Ribbon).

6/21/2021

Introduced Tectonic, a data center scale distributed file system designed to consolidate storage infrastructure. Tectonic supports exabyte-scale, isolates performance between tenants by managing resources between traffic groups, and enables tenant-specific optimizations through a client-driven microservice architecture. This consolidation has reduced data warehouse clusters by 10x, improving resource utilization and operational simplicity.

6/21/2021

Introduced Kats, a new comprehensive Python library for generic time series analysis. Kats provides algorithms and models for forecasting (including 10+ models, ensembling, meta-learning, backtesting, and prediction intervals), detection (seasonality, outliers, change points, slow trend changes), feature extraction (65+ statistical features), and multivariate analysis. It also includes utilities like time series simulators. Kats aims to standardize and connect various domains within time series analysis, enabling users to explore data characteristics, predict future values, monitor anomalies, and integrate these into ML models.

10/9/2020

Introduced Nemo, an internal data discovery engine that significantly improved data search success rate by over 50%. Nemo leverages Unicorn search infrastructure, supports sophisticated filtering and personalization, and parses natural language queries. It incorporates trust signals for data quality, recency, usage, and lineage into its ranking process. The architecture includes bulk and instant indexing, and a serving component with NLP and ML models for ranking.

10/8/2020

Introduced CG/SQL, a code generation system for SQLite that compiles stored procedures written in a T-SQL variant into C code using SQLite's C API. This system provides compile-time error checking, automatic schema management and upgrades, test code generation, and query plan analysis, aiming to simplify the creation of complex, maintainable data access layers for SQLite.

7/10/2020

This post introduces two new cryptographic protocols, Private-ID and PS3I, for private matching on set intersections. These protocols enable secure data alignment and analysis between different parties without revealing sensitive information, extending the data warehousing and analytics platform's capabilities to handle privacy-sensitive computations and large-scale data joins.

4/21/2020

Introduced Turbine, a scalable service management platform for Facebook's stream processing service. Turbine bridges the gap between general-purpose cluster management frameworks and stream processing requirements, featuring a fast task scheduler, an efficient predictive autoscaler, and a fault-tolerant update mechanism. It manages tens of thousands of machines and thousands of streaming pipelines, processing hundreds of gigabytes of data per second in real-time, balancing workload fluctuations, predicting load spikes, and completing high-scale updates efficiently.

2/6/2020

This post details advancements in ZooKeeper at Facebook, including a 10x improvement in write throughput via snapshot scheduler changes and batch commits. It also covers optimizing semaphores with a new lock API to reduce server round trips and QPS. Scalability improvements include demand control for multitenancy (data size and request rate quotas) to mitigate noisy neighbor issues, and zero-downtime migration strategies. Furthermore, the entire ZooKeeper fleet has been containerized, moving away from bare-metal deployments, simplifying fleet management and deployments.

10/7/2019

This post details the architecture and evolution of Scribe, a distributed, buffered queueing system designed to transport petabytes of logs per hour from millions of machines. It covers the system's design decisions, including simplicity, availability, scalability, and multitenancy, and explains its components like Scribed (local daemon), Write Service, and its reliance on LogDevice for durable storage, replacing HDFS. The post also discusses the trade-offs in log delivery guarantees and the evolution from Scribe's first incarnation to its current, simplified production version.

9/30/2019

Introduced MaRS (Machine-Augmented Automatic Review System) and Logical Changesets (LoChas) to efficiently ingest and maintain up-to-date OpenStreetMap (OSM) data for Meta's applications. LoChas break down OSM changesets into individual CRUD operations and cluster them for efficient human review. MaRS uses heuristics and ML to automate the evaluation of LoChas, reducing the workload requiring human intervention. This system ensures data freshness and correctness by processing changes in an idempotent, commutative, and independent manner, allowing for parallel processing and binary accept/reject decisions.

8/20/2019

This post details the technical redesign of Meta's systems to provide users with control over Off-Facebook activity. It describes the challenges of storing and processing this data, which was primarily in the data warehouse, and the need to move away from time-based partitioning to user-centric organization for this specific feature. Key technical solutions include the introduction of Separable Identifiers (SIDs) to decouple user IDs (UIDs) from off-site activity data in the data warehouse, the use of Encrypted IDs (EIDs) for compound data types, and Measurement IDs (MIDs) to enable aggregated measurement reporting while respecting user privacy. The post also outlines strategies for handling special cases, such as ensuring data security and safety investigations, and managing Facebook Login integrations.

7/9/2019

Introduced OIL (Output-Input Language) as a unified namespace and abstraction for files across heterogeneous storage systems, allowing configuration of I/O trade-offs via Directed Acyclic Graphs (DAGs) instead of coding. Coupled OIL with VCache, a distributed virtual memory system, to provide significant efficiency and performance gains. OIL extends POSIX APIs to better capture distributed system nuances, enabling per-file or per-I/O trade-off adjustments and allowing files to retain the same name even when migrating storage systems. VCache integrates with OIL to provide layered caching, supporting various write modes and delegation for buffering and durability. The system uses DAGs for both data and metadata, with the metadata DAG handling mutual exclusion and locking, and the data DAG managing byte-stream operations. OIL+VCache has been deployed to enhance live video viewer systems, improving reliability and latency while reducing computational overhead.

6/10/2019

This post details the Aria Presto initiative to improve PrestoDB efficiency, focusing on optimizing table scans for ORC format. Key technical contributions include subfield pruning for complex data types, adaptive filter ordering to reduce CPU cycles, and efficient row skipping to minimize data reading. The new architecture shifts filter evaluation to the Hive connector and introduces stream readers and record readers to manage filtering and column scanning dynamically. These optimizations aim for a 2-3x decrease in CPU time for Hive queries.

6/6/2019

Introduced Delos, a new architecture for replicated storage systems designed for control plane services. Delos features a modular, layered design separating API materialization, update ordering (VirtualLog/SimpleLogs), and reconfiguration. This allows for fast deployment (8 months for initial version), rapid evolution (10x latency reduction via hot-swapping ordering mechanisms), and flexibility in API and performance/reliability guarantees. It uses a VirtualLog abstraction that chains SimpleLogs, enabling dynamic switching between different ordering mechanisms (e.g., ZooKeeper-based, native converged, disaggregated LogDevice) based on latency SLAs.

4/25/2019

Introduced F14, a 14-way probing hash table implementation within the Folly C++ library, designed for improved performance and memory efficiency over previous specialized hash table implementations. F14 utilizes vector instructions (SSE2/NEON) for parallel intra-chunk searching and a novel reference-counted tombstone strategy (overflow bits) to handle collisions and erasures efficiently. It offers multiple memory layouts (F14FastMap/Set, F14NodeMap/Set, F14Vector) to cater to different use cases, particularly optimizing for scenarios without long-lived references to entries.

3/1/2019

Introduced LtHash, a homomorphic hashing algorithm based on lattice cryptography, for secure and scalable update propagation in configuration management systems. This addresses the challenges of maintaining data integrity across a large network of peers by enabling efficient recomputation of database hashes after updates, without iterating over the entire database. The implementation is open-sourced in Folly.

12/19/2018

This post details the integration and benefits of Zstandard (zstd) compression at Facebook, focusing on five key areas: large payload backups and package distribution (introducing multithreading, long range mode, and automatic level determination), data warehousing (increasing block size and reducing data transformations), compressed filesystems (SquashFS and Btrfs for faster deployments and tighter storage), and databases (hybrid compression with RocksDB). These enhancements have led to significant improvements in compression ratios, decompression speeds, and reduced resource consumption across these systems.

12/13/2018

This post details the implementation of the HyperLogLog (HLL) algorithm in Presto to significantly improve the performance of cardinality estimation for large datasets. It explains the evolution of cardinality estimation algorithms from simple minimum value estimation to probabilistic counting and LogLog, culminating in HLL. The post highlights the memory efficiency and speed improvements (7x to 1,000x) achieved by using HLL in Presto for tasks like analyzing A/B tests.

11/13/2018

This post summarizes presentations from the Data @Scale conference in Boston. Key technical contributions include: discussions on protecting patient privacy while using large-scale health care data sets; balancing flexibility and control in database deployments; lessons learned from scaling a timeseries database (InfluxDB), including failure conditions and trade-offs between monolithic and service-oriented implementations; leveraging sampling to reduce data warehouse resource consumption and manage uncertainty propagation in aggregated metrics; exploring Transient Replication and Cheap Quorums in Apache Cassandra for disk space and compute savings; detailing Facebook's Deletion Framework for managing data deletion at scale across distributed systems; describing Wayfair's transformation of data plumbing components into a scalable infrastructure, including the development of Tremor as a replacement for logstash; presenting Kubeflow for portable machine learning on Kubernetes, addressing scalability, portability, and composability; outlining DataXu's journey to a cloud-native warehouse using AWS services and spot instances; detailing HubSpot's process for migrating Elasticsearch instances at scale to enhance security and reduce migration time; presenting Presto's Cost-Based Optimizer (CBO) for improved join efficiency and a new mechanism for seamless statistics collection; introducing Palisade and Akamill for overload protection and stream processing in data analytics and storage systems; and discussing best practices for building highly reliable data pipelines at Datadog, including ephemeral clusters, job isolation, and rapid recovery mechanisms.

10/8/2018

Introduced Akkio, a new data placement service (DPS) designed to optimize data retrieval speed and reduce storage and network costs in geo-distributed data centers. Akkio operates at a microshard granularity, determining optimal data placement based on access patterns and available capacity. It addresses the limitations of traditional caching for Facebook's workloads by handling strong consistency requirements and high write volumes more effectively. The system leverages ZippyDB for storing access patterns and metadata, and serializes migrations through locks to prevent concurrent operations.

9/21/2018

This post announces the open-sourcing of LogDevice, a distributed data store for sequential data developed and deployed within Facebook. LogDevice provides durability, high availability, and total order of records under failures, making it suitable for workloads like event streaming, replication pipelines, transaction logs, and deferred work journals. It also introduces LDShell, a command-line administration tool.

7/19/2018

Introduced Location-Aware Distribution (LAD), a new peer-to-peer system for distributing configuration changes to millions of servers. LAD improves upon the previous ZooKeeper-based system by supporting larger updates (100 MB vs 5 MB), scaling to 40,000 subscribers per distributor (vs 2,500), and reducing distribution latency. LAD separates data storage from distribution, utilizes a tree-based peer-to-peer network for efficient updates, and employs sharding to manage metadata distribution. Lessons learned emphasize the critical role of tooling, monitoring, failure injection testing, and continuous testing for operating and maintaining large-scale P2P systems.

6/26/2018

This post details the migration of Messenger's storage system from HBase to MyRocks, integrated with RocksDB as a MySQL storage engine, and leveraging Facebook's Lightning Server SKU with NVMe flash storage. It describes the challenges of migrating petabytes of data for over a billion users without disruption, including redesigning the data schema, ensuring data consistency through dual-writing and validation, and implementing two distinct migration flows (normal and buffered) to handle different account types and traffic loads. The migration resulted in a 90% reduction in storage consumption, 50x lower read latency, and improved system resiliency and disaster recovery capabilities.

9/25/2017

This post details the successful migration of Facebook's UDB tier from the InnoDB storage engine to MyRocks, a RocksDB-based engine for MySQL. The migration achieved a 50% reduction in storage usage by addressing the space-bound nature of InnoDB on flash storage. Key technical contributions include the development and application of comprehensive data consistency verification tools (primary/secondary key checks, cross-instance checks, shadow query replay), a phased migration strategy leveraging MyRocks' ability to replicate from InnoDB and vice-versa, and addressing technical challenges like buffered I/O and slow file deletion on flash storage. The post also highlights lessons learned regarding migration time estimation, understanding service usage for feature prioritization, starting small with new software deployments, and the benefits of open-source collaboration.

8/31/2017

Introduced LogDevice, a distributed data store specifically designed for logs. LogDevice provides highly available and durable record storage with repeatable total ordering, addressing challenges in data pipelines, indexing, and distributed transactions at Facebook's scale. Key features include record-oriented append-only storage with trimming, non-deterministic record placement for improved write availability and load balancing, and a sequencer-based architecture that decouples record ordering from storage.

6/15/2017

This post recaps the Data @Scale 2017 conference, highlighting discussions on building large-scale storage systems and analytics. Key technical contributions and discussions included: Facebook engineer Pieter Noordhuis's insights on 'Accurate, Large Minibatch SGD: Training ImageNet in 1 Hour' for efficient deep learning at scale; Microsoft's Rimma Nehme discussing the next generation of globally-distributed databases; Alexey Milovidov from Yandex detailing ClickHouse, a DBMS for interactive analytics at scale; Yongsheng Wu from Pinterest sharing insights on the evolution of storage and serving; Maxim Fateev from Uber presenting Cadence, a micro-service architecture beyond request/reply; Kapil Surlaker from LinkedIn explaining UMP and XLNT for reporting and experimentation; Sergey Melnik from Google discussing Spanner's SQL evolution, including distributed query execution and storage formats; Doug Burger from Microsoft on architectures for cloud specialization with programmable hardware; and Steve Stroiney from Facebook describing the system for bulk data movement serving global data storage and processing.

5/5/2017

This post announces the rebuilding of the OnlineSchemaChange (OSC) tool in Python, replacing the original PHP version. The new Python version offers improved ease of use with a standalone CLI, better testability through a more flexible code structure and integration testing framework, and enhanced reliability with data consistency checks. Future plans include triggerless schema changes and utilizing native online DDL.

3/8/2017

This post details the validation of various software-defined storage solutions (NetApp ONTAP Select, IBM Spectrum Scale, Excelero NVMe over Fabrics, Weka.IO, Hedvig, Red Hat GlusterFS, and Ceph) on open hardware platforms like Leopard servers and Open Vault JBODs within the Disaggregate Lab. It highlights the growing software ecosystem for open hardware and its appeal to companies of all sizes.

2/7/2017

This post details the migration of a large-scale, complex (100+ stage) N-gram language model training pipeline from HQL over Hive to Apache Spark. It highlights the use of Spark's imperative DSL and RDD interface to gain finer control over data distribution, avoid data skew, and implement application-specific optimizations. Key challenges addressed include managing large '0-shards' for smoothing, potential job instability due to data skew, and executor/shuffle service OOM errors. The solution involved a progressive sharding approach, iteratively refining data distribution based on N-gram history length (1-word, 2-word, etc.) and dynamically adjusting shard sizes to ensure balanced workloads. The use of Spark's `pipe()` operator facilitated the integration of existing C++ binaries, enabling a smooth transition without re-implementing core logic.

2/3/2017

This post details the development and open-sourcing of Beringei, a high-performance in-memory time series storage engine designed to address the scaling limitations of HBase for Facebook's monitoring infrastructure. It introduces a lossless streaming compression algorithm that achieves over 90% compression for time series data by using delta-of-delta encoding for timestamps and XOR-based compression for values. Beringei offers very fast write rates and low read latencies, storing up to 10 billion unique time series and serving 18 million queries per minute, powering Facebook's real-time performance and health monitoring.

10/28/2016

This post details the ORC (ORC Restore Coordinator) system, a component of Meta's continuous MySQL backup validation infrastructure. It describes the architecture of ORC, including the Warchief (load balancer), ORC DB (state management), and Peons (restore workers). The post elaborates on the five stages of a restore job within a peon: SELECT, DOWNLOAD, LOAD, VERIFY, and REPLAY. A significant technical contribution is the detailed explanation of the binlog selection logic, which leverages Global Transaction Identifiers (GTIDs) to accurately determine and replay transactions for point-in-time restores, overcoming challenges with primary/secondary backup sources and ensuring data integrity.

10/19/2016

This post provides a quantitative and qualitative comparison between Apache Giraph and Spark GraphX for large-scale graph processing at Facebook. It details the methodology, experiment setup, and key findings regarding performance, scalability, and usability. The study reveals that Giraph can process significantly larger graphs and performs better and more efficiently than GraphX for production workloads, while GraphX offers some advantages in ease of development for certain preprocessing tasks.

9/1/2016

Introduced Zstandard, a new compression algorithm that offers smaller and faster compression, twice the decompression speed, and higher compression ratios, leading to faster data transfer and smaller storage requirements. Announced MyRocks, an open-source MySQL storage engine built on RocksDB, which reduces storage requirements by 50% in production user database tiers. Developed a 360 video stabilization technology using a combination of 2D and 3D stabilization methods with a new 'deformed rotation' motion model, improving video quality, reducing bit rate, and enabling fast upload and playback. Presented BrowserLab, an automated testing environment for detecting regressions as small as 20ms, handling dynamic content, isolating server variance, and accounting for non-deterministic code. Discussed Infer, an open-source static analyzer for Java, C, and Objective-C, with ongoing efforts to add support for C++.

8/31/2016

This post details the migration of a large-scale Hive-based feature preparation pipeline for entity ranking to Apache Spark. It highlights the challenges of the previous sharded Hive implementation (3-day runtime, difficult monitoring) and the technical solutions implemented to run a single Spark job on 60 TB of compressed data with a 90 TB shuffle. Key contributions include reliability fixes for node reboots (PipedRDD robustness, configurable fetch failures, restartable shuffle service), addressing driver issues (O(N^2) operations, excessive speculation), and executor OOM fixes. Performance optimizations involved memory leak fixes in the sorter and Snappy optimizations. The post also details the tools used for performance bottleneck identification (Spark UI Metrics, Jstack, Perf/Flame Graph) and emphasizes the contribution of these improvements back to the open-source Apache Spark project.

8/31/2016

Introduced MyRocks, a new open-source MySQL storage engine that integrates RocksDB as a backend. MyRocks offers significant improvements in space efficiency (50% less storage compared to compressed InnoDB) and reduced write amplification by leveraging RocksDB's LSM-tree structure, append-only nature, and prefix key encoding. Benchmarks show MyRocks is two times smaller than compressed InnoDB and writes orders of magnitude less. The post also details the migration process from InnoDB to MyRocks in production without downtime.

6/16/2016

This post recaps the Data @Scale conference, highlighting discussions on exabyte-scale storage systems (Dropbox), intrinsic analytics for storage visibility (Qumulo), MPP shared-nothing databases on flash (Presto Raptor at Facebook), lessons from studying modern databases and key-value stores (University of Wisconsin-Madison), flexible consistency levels with Manhattan (Twitter), scale-out resource management (YARN++ at Microsoft), a unified model for batch and streaming data processing (Apache Beam at Google), data integration for visual analysis (Tableau), and disaggregated storage and compute for Map/Reduce (Facebook).

3/18/2016

Introduced Dragon, a distributed graph query engine, to address the challenges of high-volume, multi-hop queries on Facebook's social graph. Dragon builds upon TAO by creating various indices to improve the efficiency of fetching, filtering, and reordering data, reducing network transfer and improving latency. It leverages techniques like socially aware inverted indices and optimized storage layouts based on query patterns, and supports functional programming primitives for complex filtering and ordering. Dragon aims to serve queries from cache with high availability and consistency, allowing applications to focus on business logic.

12/28/2015

This post introduces the implementation of a write-ahead log for software RAID 5/6 to address the 'write hole' problem, improve rebuild performance, and optimize write performance. The solution involves writing data and parity to a log on an SSD before persistent writes to RAID disks, ensuring consistency after power failures and enabling aggregation of writes to avoid read-modify-write penalties. The implementation details include metadata structures, checksum mechanisms, and reclaim strategies. Performance testing with different disk configurations is presented, along with future work on NVRAM/NVDIMM support and further write performance improvements.

11/5/2015

This post details the creation and open-sourcing of Presto, a distributed SQL engine designed for interactive analytic queries against large datasets. It highlights Presto's architecture, its significant query performance improvements over traditional MapReduce jobs, and its adoption by various companies like Airbnb, Dropbox, and Netflix. The post also discusses the growth of the Presto community, with an increase in contributors and support from cloud platforms and enterprise vendors.

7/31/2015

This post introduces the adoption of FlatBuffers as a data storage format for Facebook's Android application. It details the performance issues encountered with JSON parsing, such as slow parsing speeds, parser initialization delays, and significant garbage collection pressure. The post explains how FlatBuffers, with its ability to directly access data without full deserialization, addresses these issues. It also describes custom mutation extensions built on FlatBuffers to handle data modifications and the resulting 'Flat Models' architecture that simplifies the integration between UI and storage layers. The post quantifies the performance improvements achieved, including reduced story load times, decreased memory allocations, faster cold starts, and smaller storage size.

6/22/2015

This post summarizes presentations from the Data@Scale 2015 event, highlighting advancements in distributed data storage and processing. Key contributions include details on Presto (an open-source distributed SQL query engine for ad-hoc analysis), LogLens (Twitter's real-time log indexing and search service), Firefly (Dropbox's index and search system), Précis (Dato/GraphLab's scaling and resource utilization system), Azure DocumentDB (Microsoft's cloud-based NoSQL database), ZippyDB (Facebook's key/value memory cache service), Outlook Service (Microsoft's cloud-scale service architecture), Facebook's multi-exabyte cold storage systems (using HDDs and Blu-ray discs), Windows Azure Storage's media type evolution, and Backblaze Vaults (a distributed storage system).

6/2/2015

This post details the development of a distributed algorithm for collaborative filtering (CF) to handle Facebook's massive datasets (100 billion ratings, over a billion users, millions of items). It addresses the limitations of standard distributed matrix factorization approaches by introducing a rotational hybrid approach that leverages worker-to-worker messaging in Apache Giraph. This new approach significantly reduces network traffic and eliminates skewed item degree problems, improving scalability for recommendation systems.

4/30/2015

This post details how osquery utilizes RocksDB as its embedded persistent key-value store. It explains osquery's functionality as an OS instrumentation framework and how RocksDB enables fast, persistent data storage and access. The article highlights the event-based pub/sub system within osquery, demonstrating how RocksDB is used to store and compare query results over time to detect state changes. It also introduces the concept of event publishers and subscribers, using udev events as a concrete example of how osquery reacts to OS-level events efficiently by leveraging RocksDB for state management.

4/30/2015

This post details Parse's adoption of RocksDB as a storage engine for MongoDB in production. It highlights significant performance gains, particularly in write operations (50x faster inserts) and data compression (70%-90%), which allows for more indexes to fit into RAM, leading to faster overall performance. This integration enhances MongoDB's existing strengths in failover and workload flexibility.

3/17/2015

This post details the development and integration of a new ORC reader for Presto, designed to significantly improve data processing performance. Key contributions include enabling columnar reads, implementing predicate pushdown by leveraging ORC's min/max statistics, and introducing lazy reads to optimize data access based on query predicates. The new reader supports both ORC and DWRF formats and demonstrates substantial speedups in raw decoding and end-to-end query latency compared to previous implementations and other data formats like RCFile and Parquet.

1/16/2015

Alibaba's MySQL team has joined the WebScaleSQL project, contributing bug fixes, performance improvements, enhanced large-scale parallelism, and micro-optimizations in compression. This collaboration is expected to accelerate progress on planned replication and GTID server-side fixes, async client features, logical read-ahead, query throttling, improved server-level thread pooling, large InnoDB page support, and memory usage reduction.

11/19/2014

This post details presentations from the Data @Scale – Boston conference, highlighting advancements in large-scale data processing and systems engineering. Key technical contributions include: Twitter's use of the Lambda architecture and probabilistic algorithms for real-time high-volume analytics processing; Facebook's technical challenges and infrastructure enablers for the 'Lookback videos' project, focusing on compute, network, storage, distribution, projections, and modeling; Vertica's Database Designer (DBD) for automated projection design in columnar databases and Live Aggregate Projections (LAPs) for faster querying through pre-aggregation; Wayfair's scaling of Redis and Memcached using composable tools like Ketama and Twemproxy for a resilient distributed caching system; TripAdvisor's lessons learned in handling operations at scale with Hadoop, log shipping, and anomaly detection; Facebook's implementation of spatial indexing on RocksDB for efficient geo-spatial data storage and optimization for various workloads; Facebook's use of graph partitioning for optimizing distributed systems serving graph-based datasets; Constant Contact's scaling strategies using Cassandra for key-value data and sharded MySQL for relational data, and their analytics platform leveraging Hadoop; and VoltDB's architecture for high-volume transactional and stream processing.

11/10/2014

Introduced the Audience Insights (AI) query engine, a system designed for rapid processing of social analytics data. Key technical contributions include a hybrid in-memory/flash disk integer store, a fan-out distributed architecture with aggregation and leaf tiers, sharding by user with 1024 shards, and the use of bitset-based inverted indices for efficient filtering (replacing user IDs with bitmasks for faster bitwise AND operations). Implemented cumulative bitmap indices for fast range filters using XOR. Addressed memory constraints by using bitmaps for dense attributes and traditional indexing for sparse attributes, converting sparse indices to bitmaps at query time. Optimized group-by operations using bitset-based filtering and converted 64-bit integer entity IDs to 32-bit vector indices. Improved cache efficiency by organizing user attributes in a columnar-like structure. Scaled affinity computation using greedy selection over locally computed candidates at each shard. Leveraged GPU hardware for offloading computationally intensive operations like affinity computation. Implemented tail latency reduction techniques by cutting execution for high-latency nodes. The engine is implemented in C++ and uses internal Facebook frameworks.

10/21/2014

This post details the Data track from the @Scale 2014 conference, highlighting advancements in real-time analytics for mobile products at Facebook, Netflix's data platform architecture centered around S3 and stream processing with Mantis, Pinterest's graph storage service (Zen), Facebook's open-sourced mcrouter for scaling cache deployments, Box's approach to structured data and enterprise cloud infrastructure with Tron and Credence, Facebook's Deltoid for real-time A/B test analysis, Facebook's mobile-first backend for Messenger (Iris) using MySQL with advanced features, and YouTube's Vitess for scalable MySQL-based data storage.

9/18/2014

This post details the deployment of MySQL's Global Transaction ID (GTID) feature across Facebook's production MySQL instances. It highlights the benefits of GTID for failover, point-in-time backup recovery, and hierarchical replication. The post outlines significant challenges encountered during deployment, particularly with existing high-availability strategies, and describes custom solutions developed in fb-mysql to overcome these. These include relaxing gtid_mode constraints during deployment, fixing bugs related to binary log scanning and crash-safety with less durable settings, and enabling multi-threaded secondary replication with relay_log_recovery. It also covers preparation steps like auditing applications for GTID-incompatible statements and developing helper scripts for statement skipping. The work done is planned to be open-sourced via WebScaleSQL.

6/5/2014

Introduced HydraBase, a system that enhances HBase reliability by using a quorum-based approach with RAFT consensus for region hosting. This allows for faster failover and higher availability compared to traditional HBase primary-secondary setups, by having multiple region servers synchronously write to the WAL and tolerating up to F failures in a 2F+1 quorum. HydraBase enables failover at the granularity of a single region and reduces failover time by eliminating log splitting.

6/5/2014

This post details the implementation and lessons learned from deploying HDFS RAID (Erasure Codes) to reduce HDFS replication factors, leading to tens of petabytes of capacity savings. It introduces XOR and Reed-Solomon erasure codes, explains the 'small file problem' and the solution of 'directory RAID', and discusses challenges like data corruption, parallelizing RAID for large directories, and handling directory changes. Future work includes native HDFS RAID support.

4/10/2014

This post details the evolution of Facebook's Hive storage format from RCFile to ORCFile, driven by the need for improved storage efficiency for their 300 PB data warehouse. Key contributions include the design and implementation of adaptive column encodings (dictionary encoding for strings and integers based on observed values and character sets), optimizations for write performance (using hash maps for dictionaries, reducing dictionary sorting overhead, and intelligent skipping of encoding for subsequent stripes), and improvements to read performance through lazy column decompression and decoding leveraging index strides. These changes resulted in a 5x to 8x compression ratio improvement over RCFile and a 3x average improvement in write performance compared to open-source ORCFile.

3/27/2014

This post announces WebScaleSQL, a collaboration among Facebook, Google, LinkedIn, and Twitter to build upon the upstream MySQL branch. Key contributions include an automated framework for running and publishing MySQL's built-in test system (mtr) results, a new suite of stress tests and a prototype automated performance testing system, changes to existing tests and code structure to improve maintainability, performance improvements such as buffer pool flushing enhancements and query optimizations, and new features like `super_read_only` and sub-second client timeouts. Future work includes contributing an asynchronous MySQL client, moving production-tested table, user, and compression statistics into WebScaleSQL, and adding the Logical Read-Ahead mechanism.

11/21/2013

This post introduces RocksDB, an embeddable, persistent key-value store developed and open-sourced by Meta. It addresses the challenge of storing and accessing hundreds of petabytes of data efficiently for real-time, user-facing applications. RocksDB is built on LevelDB and is designed to scale on multi-core servers, efficiently utilize fast storage (like flash), offer a flexible architecture for innovation, and support various workloads (IO-bound, in-memory, write-once). It aims to reduce network latency by enabling applications to manage their own datasets locally and improve performance by minimizing lock contention and optimizing storage operations.

11/6/2013

Introduced Presto, a new distributed SQL query engine optimized for ad-hoc analysis at interactive speed, designed to address the limitations of Hive/MapReduce for petabyte-scale data warehouses. Presto employs an in-memory, pipelined execution model that avoids disk I/O between stages, significantly reducing query latency. It supports standard ANSI SQL and is implemented in Java with dynamic bytecode compilation for performance. Presto's architecture includes a storage abstraction with connectors for querying disparate data sources like HDFS, HBase, and custom systems. The system was deployed company-wide, scaled to 1,000 nodes, and is used by over a thousand employees, processing one petabyte daily, achieving 10x better CPU efficiency and latency than Hive/MapReduce for most queries. Future roadmap includes removing join/aggregation size restrictions, enabling output table writes, and developing a query accelerator with a new data format. Presto is also being open-sourced.

10/24/2013

This post details the engineering challenges and solutions for building a posts search index capable of handling trillions of posts. Key contributions include: collecting data from a varied MySQL schema, indexing data in HBase using Hadoop MapReduce, updating the index via a pub/sub system (Wormhole) that bypasses caches for initial collection but hits them for updates, serving a large index from a combination of RAM and SSDs, and implementing a sophisticated ranking system with query rewriting and dynamic result scoring based on over a hundred features. The project originated from a hackathon and involved significant infrastructure development.

10/22/2013

Introduced MySQL Pool Scanner (MPS), a state machine system written in Python to automate routine MySQL DBA tasks for Meta's large-scale database clusters. MPS manages instance states (production, spare, drained, etc.) and uses 'problems' (tags) to drive automated actions like instance replacement and primary promotion, enabling bulk maintenance operations with minimal human intervention. It relies on a central repository for state management and integrates with Xtrabackup for data copying and Windex for reimaging.

8/14/2013

This post details the significant enhancements made to Apache Giraph to scale it to handle trillions of edges. Key contributions include: flexible vertex/edge based input allowing loading from separate sources and multiple Hive tables; HiveIO for up to 4x faster reading/writing to Hive tables; multithreading for graph loading, computation, and result storage; memory optimizations via serialization of vertices/edges into byte arrays and using Java primitives with FastUtil; and sharded aggregators implemented with Netty to bypass Zookeeper limitations and handle large-scale global computations efficiently.

6/25/2013

Introduced TAO (The Associations and Objects), a graph-based data storage system designed to handle Facebook's demanding social graph workload. TAO provides a unified API for managing objects (nodes) and associations (edges), abstracting away the complexities of distributed caching (Memcache) and persistent storage (MySQL). It addresses limitations of client-side graph APIs by implementing a distributed service for efficient cache management, consistency, and reduced network overhead. TAO partitions data into shards, uses a two-tier caching system (leaders and followers), and optimizes for creation-time locality.

4/1/2013

Introduced LinkBench, a new database benchmark designed to accurately simulate Facebook's social graph workload. LinkBench replicates the data model, graph structure, and request mix of MySQL social graph data, focusing on transactional workloads for serving real-time queries and updates. The benchmark was developed by analyzing production workload traces to understand out-degree distributions, read/write access patterns (power-law distributions), and the impact of high-degree nodes. It includes a Java driver for generating graphs and operations, with phases for loading and requesting data, and provides adapters for different database systems. Benchmark results for MySQL 5.1.53 are presented, showing throughput and latency under load, along with analysis of server resource utilization.

3/6/2013

This post details the development of Unicorn, a unified, scalable inverted-index search engine designed to support Graph Search. It describes the migration of legacy search systems (PPS, Typeahead) to Unicorn, the challenges of indexing the Facebook graph (nodes and edges with metadata), and the design of Unicorn's index structure (posting lists, static rank) and query language. It highlights Unicorn's ability to handle multi-hop queries and the real-time update pipeline for indexing over 2.5 billion new pieces of content daily.

3/5/2013

This post introduces McDipper, a highly performant flash-based cache server that is Memcache protocol compatible. It addresses the challenge of efficiently caching large datasets with moderate to low request rates by leveraging flash storage. McDipper is presented as a drop-in replacement for Memcached, offering improved capacity and performance. The post details the technical challenges in achieving feature parity with Memcached, including handling race conditions between sets and invalidations, and introduces solutions like delete hold-offs and read-modify-write operations. A significant application of McDipper is highlighted in Facebook's photo infrastructure, where it serves over 150 Gb/s from forward caches in the CDN, drastically reducing server counts and improving latency. The post also emphasizes the importance of configurability and collaboration in the successful deployment of such infrastructure.

1/14/2013

This post details the implementation of a three-stage automated backup system for Facebook's large-scale MySQL installations. Stage 1 involves rack backups (RBUs) collecting binary logs and using mysqldump for nightly logical backups. Stage 2 utilizes Hadoop DFS for stable, replicated storage of these backups. Stage 3 involves weekly copying to discrete storage in a separate region. The post also describes a novel exponential scoring system for backup failure severity monitoring and a continuous restore testing process with a self-service restore coordinator (ORC).

11/8/2012

Introduced Corona, a new scheduling framework for MapReduce jobs that addresses limitations in Hadoop's original scheduling framework. Corona separates cluster resource management from job coordination by introducing a cluster manager and dedicated job trackers per job. It uses a push-based scheduling model to minimize latency and implements fair-share scheduling for better fairness guarantees and multi-tenant support. The deployment involved a staged rollout across the production cluster, migrating workloads team by team. Key benefits include improved slot refill times (17% improvement in production, 600ms in simulation), higher cluster utilization (over 95% vs 70%), reduced unfairness (14.3% to 3.6%), and halved job latency for a test job. Future work includes resource-based scheduling and integrating other applications like Peregrine.

6/13/2012

This post details the development and open-sourcing of Avatarnode, a highly available Namenode solution for Hadoop Distributed Filesystem (HDFS). It addresses the single point of failure (SPOF) issue with the traditional Namenode, which was responsible for 41% of HDFS-related incidents in Facebook's Data Warehouse. Avatarnode uses ZooKeeper to manage failover between a Primary and Standby Namenode, ensuring continuous operation and reducing planned downtime. The post also highlights the significant load on the Namenode in large-scale HDFS deployments and the impact of Namenode unavailability on cluster functionality.

3/4/2012

This post details Mark Callaghan's work on optimizing MySQL for Facebook's scale, focusing on challenges with multi-core servers and fast storage. Key contributions include adapting innoDB compression for write-heavy workloads and faster storage, improving performance bottlenecks in the database server (e.g., evictions per second), and optimizing checksum algorithms (e.g., using CRC32) with on-the-fly upgrades.

12/29/2011

This post details the engineer's work on a distributed SQL database designed to scale beyond single-machine limitations and handle diverse data access patterns, aiming to address future challenges in managing large datasets.

7/27/2011

This post details a large-scale data migration of dozens of petabytes of Hadoop data between data centers. It describes the decision to use a replication system over a physical move to minimize downtime. The replication process involved a bulk copy using DistCp with custom modifications for Facebook's dataset (handling multiple mappers for large files and directories with many small files) followed by a custom Hive plug-in to record changes in an audit log, which a new replication system then polled to copy modified files and metadata. The switchover involved stopping the JobTracker, allowing replication to catch up, and then updating DNS entries. Challenges included developing a multi-threaded replication system capable of handling millions of objects and managing the numerous systems interacting with the MapReduce cluster during the switchover. The replication system also demonstrated potential for disaster recovery.

5/5/2011

This post details a collaboration with Harvey Mudd College undergraduates to improve indexing support in Apache Hive. The students added support for automatic index usage and a new bitmap index type, which was tested with encouraging results and is being prepared for testing on production clusters. The work also involved contributions to open source collaboration and leveraged existing research on bitmap indexes.

3/15/2011

Introduced a new real-time analytics system for website insights, capable of processing over 20 billion events per day with sub-30-second latency. The system architecture leverages HBase for high write throughput and reliability, scribe for event logging atop HDFS, and a client for batching and writing data streams. Explored and rejected MySQL counters, in-memory counters, and MapReduce due to write rate, reliability, and latency limitations, respectively. Discussed architectural details like table schema, key composition, batching nuances, and sharding.

3/1/2011

This post details enhancements to Facebook's database backup system, migrating from slow, logical mysqldump to a custom physical backup model using XtraBackup. Key improvements include: supporting fast table-level restore, enhancing full and incremental backups (including remote streaming), and developing a 'hybrid incremental' approach that uses a page tracker to avoid reading unmodified pages while optimizing IO calls. These changes significantly reduce backup times and space requirements.

12/15/2010

This post details optimizations to join operations in Apache Hive, a data warehouse infrastructure built on Hadoop. It introduces two key improvements: HIVE-1641, which uses the Hadoop Distributed Cache to efficiently distribute small join tables to mappers, and HIVE-1642, which automatically converts common joins to map joins based on input table sizes during execution. Performance benchmarks show significant speedups (12-26x for optimized map join, 57-163% for auto-converted joins). Future work includes dynamic replication factor determination for compressed hashtable files, smarter table size estimation using sampling, and memory-efficient hashtable implementations.

12/14/2010

This post details the process of visualizing friendship locality using data from Apache Hive. It describes sampling 10 million friend pairs, combining this data with user city information to sum friendships between city pairs, and then merging with city longitude and latitude. The post outlines the iterative process of visualization in R, including initial attempts with simple line plotting, the challenges of rendering dense data, and the eventual solution of using weighted lines based on Euclidean distance and friendship count, with a color ramp to represent weight. It also mentions replacing lines with great circle arcs for more accurate geographical representation.

12/10/2010

This post details Meta's use of Apache Hadoop in three key areas: data warehousing for web analytics, real-time application serving (with HBase and Scribe), and MySQL database backups. It announces the open-sourcing of their production-tuned Hadoop 0.20 branches. Key technical contributions and features highlighted include: enhancements to the NameNode's locking model for scaling to 30 petabytes, integration of Scribe for near real-time data streaming into Hadoop, and the implementation of reliable HDFS file-append operations for zero data loss with HBase. Specific innovations discussed are the AvatarNode for Hadoop High Availability (contributed as HDFS-976), HDFS-RAID with XOR-parity for storage reduction (contributed to Hadoop MapReduce), and the file-append feature for HBase (contributed to the 0.20-append branch). Future work includes cross-data center Hadoop functionality (HDFS-1432), resource-aware scheduling, slotless MapReduce, HDFS snapshots, and downtime-free JobTracker updates.

12/2/2010

This post contributes to the understanding of data processing and performance by introducing a comprehensive framework for analyzing system performance based on data characteristics and hardware I/O. It details the hierarchy of computer memory and storage, explaining their latency differences and impact on performance. The post also provides practical advice on performance analysis through estimation and understanding fundamental bottlenecks.

5/17/2010

This post details the engineering behind Facebook's interactive typeahead search feature, showcasing how the data warehousing and analytics platform evolved to support real-time, low-latency query processing. It describes the architecture involving front-end bootstrapping, AJAX requests, an aggregator service, and specialized leaf services (global and graph) that query indices. The post highlights the integration of caching (memcached) and various data stores (MySQL), the challenges of balancing performance, recall, and relevance, and the iterative process of testing and refinement to achieve a production-ready system for millions of users.

3/16/2010

This post details the foundational aspects and complexities of Facebook's search product, highlighting its critical role in user engagement and social graph navigation. It introduces the concept of user-centric search, emphasizing personal and social context as key relevance indicators. The post outlines significant technical challenges, including ranking on the critical path, the lack of traditional query caching due to user-specific searches, the need for a large 'hot' index, and the complexities of live data updates. It also touches upon the expansion of search beyond people to include content, bands, and applications, and the infrastructure required for indexing public and friend content.

6/10/2009

This post introduces Hive, a petabyte-scale data warehousing framework built on Hadoop, developed by Facebook's Data Infrastructure Team. It addresses the limitations of previous RDBMS-based data warehouses in handling rapidly growing data volumes (from 15TB in 2007 to 2PB in 2009). Hive provides a familiar SQL-like interface and concepts (tables, columns, partitions) for querying unstructured data in Hadoop, making it accessible to users without extensive MapReduce programming knowledge. Key architectural components include a Metastore for metadata, a query compiler and execution engine that translates SQL to MapReduce jobs, and extensibility through SerDe (Serializer/Deserializer) and UDFs (User Defined Functions). The post details Hive's data model, including partitioning and bucketing for data pruning and management, and its query optimization techniques such as predicate pushdown and column pruning. It also highlights Hive's flexibility in handling various data formats and user-defined types/functions, and its support for custom Map/Reduce scripts via SQL extensions.

10/24/2008

This post introduces Scribe, a unified system developed to address the challenge of collecting billions of messages daily from Facebook's servers for various use cases like access logs, performance statistics, and News Feed actions. Scribe was designed for scalability, flexible network topology, and a balance of reliability suitable for logging. Its simple data model (category and message) and implementation using Thrift are highlighted as key design decisions that facilitated its widespread adoption and ease of development, contributing to the broader data collection and processing infrastructure.

6/5/2008

This post details Facebook's early adoption and scaling of Hadoop for large-scale data storage and processing. It highlights the use of Hadoop's distributed file system and map-reduce paradigm to enable previously impossible projects, leading to features like the Facebook Lexicon and improved search relevance. The post describes the deployment of multiple Hadoop clusters, daily data loading volumes, and the diverse range of projects utilizing this infrastructure. Key decisions aiding adoption include language flexibility for map-reduce programs and the embrace of SQL via the in-house data warehousing layer called Hive, which offers classic data warehouse features and is planned for open-source release.