Source-linked AI summary
Understanding Data Storage and Ingestion for Large-Scale Deep Recommendation Model Training
Mark Zhao, Niket Agarwal, Aarti Basant, Bugra Gedik, Satadru Pan, Mustafa Ozdal, Rakesh Komuravelli, Jerry Pan, Tianshu Bao, Haowei Lu, Sundaram Narayanan, Jack Langman, Kevin Wilfong, Harsha Rastogi, Carole-Jean Wu, Christos Kozyrakis, Parik Pol
TL;DR
Datacenter-scale training depends on DSI pipelines to store and serve massive datasets, and the paper examines how this infrastructure affects training at Meta. It presents an end-to-end pipeline with scalable online preprocessing and characterizes its production workloads. The characterization finds that 40% of bytes contribute to over 80% of read throughput, and identifies infrastructure bottlenecks and research opportunities.
Problem
DSI pipelines are critical to training capacity and performance but have received little consideration compared with model training.
Source: §1 · pp. 1–2Method
The paper presents Meta’s production DSI pipeline and characterizes its workloads, including scalable preprocessing, collaborative training, and storage use.
Source: §1 · p. 2Results
40% of bytes contribute to over 80% of read throughput across training jobs for a given model.
Source: §5.3 · p. 8Takeaways & Limitations
The production characterization identifies DSI hardware bottlenecks and motivates research into heterogeneous hardware, datacenter scheduling, and benchmark datasets.
Source: §1 · p. 2, §9 · p. 13The authors prioritize developer productivity and retain features even at the cost of storage, limiting feature collection reductions.
Source: §5.1 · p. 7
Abstract
from arXiv · showhide
Datacenter-scale AI training clusters consisting of thousands of domain-specific accelerators (DSA) are used to train increasingly-complex deep learning models. These clusters rely on a data storage and ingestion (DSI) pipeline, responsible for storing exabytes of training data and serving it at tens of terabytes per second. As DSAs continue to push training efficiency and throughput, the DSI pipeline is becoming the dominating factor that constrains the overall training performance and capacity. Innovations that improve the efficiency and performance of DSI systems and hardware are urgent, demanding a deep understanding of DSI characteristics and infrastructure at scale. This paper presents Meta's end-to-end DSI pipeline, composed of a central data warehouse built on distributed storage and a Data PreProcessing Service that scales to eliminate data stalls. We characterize how hundreds of models are collaboratively trained across geo-distributed datacenters via diverse and continuous training jobs. These training jobs read and heavily filter massive and evolving datasets, resulting in popular features and samples used across training jobs. We measure the intense network, memory, and compute resources required by each training job to preprocess samples during training. Finally, we synthesize key takeaways based on our production infrastructure characterization. These include identifying hardware bottlenecks, discussing opportunities for heterogeneous DSI hardware, motivating research in datacenter scheduling and benchmark datasets, and assimilating lessons learned in optimizing DSI infrastructure.
1 INTRODUCTION
The DSI pipeline is a critical but understudied part of datacenter-scale training: it must store and feed massive datasets, and its design affects training performance and capacity. The paper presents Meta’s end-to-end pipeline and characterizes production DLRM workloads to identify system bottlenecks and research opportunities.
1 INTRODUCTION
- The DSI pipeline must generate samples, store exabytes of data, and feed trainers, yet has received little consideration relative to model training.The paper focuses on DLRMs because they underpin personalization and ranking services, consume much of Meta’s training cycles and DSI capacity, and pose challenges not captured by current benchmarks or systems.Source: §1 · pp. 1–2
- Storage and online preprocessing can consume more power than GPU trainers, while dataset sizes and ingestion bandwidth grew over 2x and 4x, respectively, in two years.The paper argues that these demands can constrain training capacity under fixed datacenter power budgets.Source: §1 · p. 2, Figure 1 · p. 2, Figure 2 · p. 2
- Meta’s pipeline uses a centralized warehouse and Tectonic distributed storage, alongside DPP, a disaggregated preprocessing service that scales from tens to hundreds of nodes per job.The paper also describes DLRM workloads with evolving features and samples that require substantial compute, network, and memory resources during preprocessing.Source: §1 · p. 2
- The paper characterizes collaborative DLRM training, data generation and storage, and online preprocessing on production hardware to identify bottlenecks and insights.Its stated contributions also include an end-to-end account of the production DSI architecture and open research questions for scaling it.Source: §1 · p. 2
2 RECOMMENDATION MODEL BACKGROUND
Recommendation models predict how likely a user is to interact with a candidate item, using user and item features. At Meta, these models are trained as distributed jobs across large GPU clusters and depend on DSI pipelines to supply preprocessed tensors.
2 RECOMMENDATION MODEL BACKGROUND
- Recommendation models combine user and candidate-item features to predict likely interactions, such as clicks, and rank content for users.The paper describes training these models with mini-batch stochastic gradient descent, updating weights iteratively from successive batches of samples.Source: §2 · pp. 2–3
- ZionEX nodes each contain eight A100 GPUs, with dedicated networks for inter-node training communication and data ingestion.The training nodes connect thousands of GPUs into datacenter-scale clusters.Source: §2 · p. 3
- Meta trains hundreds of recommendation models and distributes mini-batches across replicated and sharded models on multiple ZionEX nodes.Trainers synchronize embeddings, activations, and gradients over the backend network, and jobs continue until a model-quality metric is reached.Source: §2 · p. 3
- Each job’s DSI pipeline generates and stores training samples, then preprocesses them into tensors loaded into GPU memory.The paper identifies these tensors as the preprocessed form of training samples used by the model.Source: §2 · p. 3
3 DISAGGREGATED DATA STORAGE, INGESTION, AND TRAINING PIPELINE
Meta’s DSI pipeline combines data generation and storage with online preprocessing for training jobs. Its design supports evolving, feature-rich datasets and provides a scalable service to transform stored samples into tensors for trainers.
3 DISAGGREGATED DATA STORAGE, INGESTION, AND TRAINING PIPELINE
- The pipeline’s architecture distinguishes data flow from control flow, and the paper presents its storage, preprocessing, and training systems as one end-to-end design.Figure 3 uses solid and dashed lines for data and control flow, respectively.Source: §3 · p. 3, Figure 3 · p. 4
3.1.1 Data Generation.
- Streaming and batch ETL jobs join and label serving-time feature and event logs into samples and partitioned offline datasets for model training.Logging features and events at serving time helps avoid data leakage between serving and training.Source: §3.1 · p. 3, §3.1.1 · p. 4
3.1 Data Generation and Storage
- Storage must support tens of thousands of frequently changing features, selective reads, shared access across hundreds of models, and continuously updated datasets.A centralized warehouse also supports engineers’ interactive Spark and Presto queries.Source: §3.1 · p. 3
3.1.2 Data Storage.
- Hive tables store structured rows with dense and sparse features in map columns, while DWRF flattens feature columns to enable storage-layer filtering.The columnar files are stored in Tectonic, an exabyte-scale distributed append-only filesystem.Source: §3.1.2 · p. 4
3.2 Online Preprocessing
- Online preprocessing extracts and transforms stored samples into batched tensors, with transformations applied within each mini-batch.Because it is on the training critical path, preprocessing must match trainer throughput while avoiding over-provisioning and data stalls.Source: §3.2 · p. 4
3.2.1 Scalable Preprocessing with DPP.
- DPP separates control and data planes: its Master distributes dataset splits, while stateless Workers process them and serve tensors to training nodes.Workers read and decode Tectonic data, filter unused features, apply transformations, batch samples into tensors, and buffer tensors in memory.Source: §3.2.1 · pp. 4–5
- The DPP Master monitors worker utilization and buffered tensors to scale workers toward meeting trainer demand while maximizing resource utilization.It also checkpoints reader state and can restart failed Workers without restoring a checkpoint because Workers are stateless.Source: §3.2.1 · p. 5
4 COORDINATED TRAINING AT SCALE
Meta’s collaborative release process combines exploration, large concurrent combo jobs, and release-candidate training across a global fleet. Combo-job peaks, regional dataset needs, and rapid feature changes shape the capacity and scheduling requirements of the DSI infrastructure.
4.1 Collaborative Release Process
- Combo jobs combine promising ideas in many concurrent permutations, and jobs can exceed 10 days while failures and asynchronous launches create substantial temporal skew.Exploratory jobs use less compute and typically under 5% of a table; combo jobs use most of the table within a short window.Source: §4.1 · pp. 5–6
4.2 Global Training Utilization
- Global compute utilization peaks when many models run combo jobs, so datacenters must provision storage, preprocessing, and training capacity for those release-critical periods.The fleet spans regions, and the global scheduler balances jobs for each model across regions, requiring each region to hold a copy of every model’s dataset.Source: §4.2 · p. 6
4.3 Feature Engineering
- Production feature sets change rapidly, with hundreds of features added and deprecated each month, requiring storage infrastructure to adapt to evolving schemas.Features may be beta, experimental, active, deprecated, or reaped as models progress through exploration and release.Source: §4.3 · p. 6
4.4 Summary of Key Takeaways
- Across hundreds of models, combo-job peaks motivate regional capacity planning and efficient job and dataset co-location; temporal skew and evolving data challenge optimizations built for synchronized jobs or static datasets.Model compute demand varies by region, while constrained cross-region bandwidth makes local availability important.Source: §4.4 · p. 6, §4.2 · p. 6
5 UNDERSTANDING DATA STORAGE AND READING
Production recommendation training reads petabyte-scale datasets selectively by row and feature, while jobs reuse popular bytes across runs. The resulting workload combines large centralized reads, narrow column selection, and uneven data popularity.
5.1 Individual Jobs Read and Filter Large Datasets
- Production jobs select table partitions and feature projections rather than repeatedly preprocessing epochs, and even large jobs read only part of the available data.The selected partitions can still require petabytes, and production dataset sizes continue to grow.Source: §5.1 · p. 7
5.2 Data is Reused Across Training Jobs
- Across one month of runs, the most popular 39%, 37%, and 18% of RM1, RM2, and RM3 datasets, respectively, account for 80% of storage traffic.For RM1 and RM2, individual jobs read 37% and 34% of stored bytes, while jobs collectively read over 60%.Source: Figure 7 · p. 7, §5.2 · pp. 7–8
5.1 Individual Jobs Read and Filter Large Datasets
- Jobs read only 9–11% of stored features but 21–37% of stored bytes, because selected features tend to have greater sample coverage and sparse-feature lengths.The representative models require 504–1221 dense and 42–306 sparse features.Source: §5.1 · p. 7
- Feature collection is not aggressively reduced because engineers prioritize experimentation and access to needed features, even at a storage cost.Columnar storage and extensive filtering also create relatively small contiguous storage reads, making disk-seek performance important.Source: §5.1 · p. 7
5.3 Summary of Key Takeaways
- Across petabyte-scale datasets, training jobs filter both rows and features, and 40% of bytes contribute to over 80% of read throughput across jobs for a model.Even the largest jobs often read less than the full dataset and each sample only once; their selected partitions still exceed trainer-node local storage.Source: §5.1 · p. 7, §5.3 · p. 8
6 UNDERSTANDING ONLINE PREPROCESSING
Meta characterizes the throughput and resource demands of online preprocessing across recommendation models, finding that trainer-local preprocessing can stall GPUs and that data loading and extraction strain host resources. Its disaggregated DPP service offloads extraction and transformation, while resource needs and bottlenecks vary by model and operation.
6 UNDERSTANDING ONLINE PREPROCESSING
- A production RM1 training job spent 56% of GPU cycles stalled waiting for data when preprocessing ran on trainer CPUs.The measured trainer had two 28-core CPU sockets, two 100 Gbps frontend NICs, and eight NVIDIA V100 GPUs.Source: §6 · p. 8
6.1 GPU Training Throughput
- GPU tensor-ingestion requirements vary by over 6× across representative models, and are projected to increase 3.5× within two years.The variation reflects differences in operational intensity and GPU synchronization overhead; the projection reflects larger samples, improved accelerators, and software optimizations.Source: §6.1 · p. 8
6.2 Data Loading at GPU Trainers
- Loading preprocessed tensors can require up to 40% of trainer CPU cycles and 55% of memory bandwidth, while production training approaches NIC saturation.Memory bandwidth saturates at approximately 70% utilization; TLS decryption and Thrift deserialization add to network-stack and memory-management work.Source: §6.2 · p. 8, Figure 8 · p. 8
6.3 Extracting & Transforming Data at DPP Workers
- Meeting extraction and transformation demand takes 9–55 DPP servers per trainer node, and extracting raw samples requires 1.18–3.64× more network bandwidth than loading preprocessed tensors.RM2 workers approach practical ingress NIC throughput limits; worker bottlenecks also vary by model, including CPU and memory bandwidth for RM1 and memory capacity for RM3.Source: §6.3 · p. 9
6.4 Transformation Operations
- Feature generation uses around 75% of transformation cycles, compared with around 20% for sparse normalization and 5% for dense normalization.Common feature-generation operations include Bucketize, NGram, and MapId.Source: §6.4 · p. 10
6.5 Summary of Key Takeaways
- DPP offloads extraction and transformation to eliminate GPU data stalls, but trainers still need sufficient host resources for GPU-driven data loading; memory bandwidth is expected to become the primary worker bottleneck.Feature generation dominates transformation compute.Source: §6.5 · p. 10
7 KEY INSIGHTS AND RESEARCH OPPORTUNITIES
The section identifies storage, ingestion, and datacenter-planning bottlenecks, then discusses hardware and system co-design opportunities. Its optimizations improved DPP and storage throughput while reducing DSI power requirements.
7.1 Hardware Bottlenecks in DSI
- Even after triplicate replication, HDD storage has an over 8× throughput-to-storage gap, requiring excess capacity to meet IOPS demand.Storage requirements depend on dataset size, trainer throughput, preprocessing-related data-volume changes, and I/O-size effects on achievable HDD IOPS.Source: §7.1 · p. 10
7.2 Heterogeneous Hardware for DSI
- SSD nodes provide 326% IOPS per watt but only 9% capacity per watt compared with fleet HDDs, so storage media must balance throughput and capacity.Putting all training data on SSDs would create an unfavorable storage-to-throughput gap given the large datasets.Source: §7.2 · p. 10
- Transformation placement is operation- and model-dependent: on a V100 GPU versus 20 CPU threads, SigridHash achieved 11.9× and Bucketize 1.3× performance.GPU acceleration faces small, diverse operation graphs, launch and transfer overheads, variable-length outputs, and potential training-throughput degradation; TLS operations also amplify memory bandwidth by 3×.Source: §7.2 · p. 11
7.3 Datacenter Planning and Scheduling
- Planning must jointly provision compute, network, and storage for fixed power budgets and peak demand, while profiling workloads and accounting for preprocessing needs.A global scheduler could route and bin-pack jobs across regions to reduce storage duplication; model-hopper parallelism could move TB-scale models instead of PB-scale data as datasets grow beyond datacenter scale.Source: §7.3 · p. 11
7.4 Representative Benchmark Datasets
- Existing benchmarks largely emphasize models over datasets, while production recommendation workloads use continuously generated, petabyte-scale data with extensive preprocessing and filtering.These workloads use one epoch and row-wise and feature-wise filtering, motivating more representative benchmark datasets.Source: §7.4 · p. 11
7.5 Multi-dimensional System Co-design
- 2.94× higher DPP throughput and 2.41× higher storage throughput, with 2.59× lower DSI power requirements, resulted from co-designed optimizations.These power reductions allowed datacenters to provision significantly more compute resources for trainers.Source: Table 12 · p. 12, §7.5 · p. 12
- Feature flattening doubled DPP Worker throughput with a 12% storage-capacity increase; coalesced reads addressed a 97% storage-throughput loss from small filtered I/Os.Feature reordering then reduced unnecessary reads and improved storage throughput by 84%.Source: §7.5 · p. 12
- Large stripes improved storage throughput by 31%, in-memory flatmaps improved Worker throughput by 15%, and localized optimizations further improved DPP throughput by 28%.The section frames these as end-to-end and top-to-bottom co-design, including trade-offs such as storage capacity for DPP efficiency.Source: §7.5 · p. 12
8 RELATED WORK
Prior work addresses ETL, warehousing, online preprocessing, and individual DSI components, often in benchmark settings. This paper characterizes an end-to-end DSI deployment for large-scale training.
ETL pipelines
- Traditional ETL engines generate structured training data from raw logs, while online preprocessing requires PyTorch integration, pipelined compute, localized mini-batch transforms, and right-sizing.The discussion identifies a need for a distinct online preprocessing framework focused on power efficiency and avoiding data stalls.Source: §8 · p. 12
Data storage and warehousing for ML
- Hive tables on Tectonic with an optimized columnar format address industrial datasets that are massive, evolving, selectively filtered, and reused across models and systems.The passage also notes feature stores and data warehouses as comparable industry solutions.Source: §8 · pp. 12–13
Online preprocessing for ML
- DPP is a fully managed, disaggregated online-preprocessing runtime integrated with PyTorch, whereas tf.data focuses on host-CPU preprocessing.DPP automatically scales to meet throughput needs of jobs running on hundreds of GPUs.Source: §8 · p. 13
Other recent works and large-scale training
- Industrial DSI characteristics differ markedly from benchmarks, limiting the impact of systems designed for benchmark vision and NLP workloads in industrial settings.Prior work targets caches, randomized minibatches, sample augmentation, TPU data stalls, and hyperparameter-tuning preprocessing.Source: §8 · p. 13
9 CONCLUSION
The paper presents Meta’s end-to-end data storage, ingestion, and training pipeline for production recommendation models and characterizes its workloads and infrastructure. It synthesizes research directions from that characterization and experience.
9 CONCLUSION
- The paper presents Meta’s end-to-end data storage, ingestion, and training pipeline used for production recommendation models.It characterizes coordinated training, dataset storage and reading, and online preprocessing workloads, then synthesizes research directions.Source: §9 · p. 13