Source-linked AI summary
Column-Oriented Storage Techniques for MapReduce
Avrilia Floratou, Jignesh Patel, Eugene Shekita, Sandeep Tata
TL;DR
As Hadoop workloads scale, performance issues arise in a system whose programming and storage constraints complicate the use of parallel-DBMS techniques. The paper adapts column-oriented binary storage to Hadoop while preserving its APIs, and reports performance benefits alongside trade-offs for smaller datasets.
Problem
Method
The paper introduces a column-oriented binary storage format for Hadoop, with lightweight compression and lazy record construction to avoid deserializing unwanted records.
Source: §1.1 · p. 2, §8 · p. 8Results
CIF offers flexibility and performance benefits over RCFile, including no performance penalty for that flexibility and improvements to metadata overhead, prefetching, and I/O elimination.
Source: §4.3 · p. 4Takeaways & Limitations
CIF can make adding columns less costly than with RCFile, where adding a column requires reading and rewriting the entire dataset.
Source: §4.3 · p. 4CIF requires a special HDFS block-placement policy and may limit parallelism for smaller datasets, though the paper does not expect these issues to affect large deployments.
Source: §4.3 · p. 5
Abstract
from arXiv · showhide
Users of MapReduce often run into performance problems when they scale up their workloads. Many of the problems they encounter can be overcome by applying techniques learned from over three decades of research on parallel DBMSs. However, translating these techniques to a MapReduce implementation such as Hadoop presents unique challenges that can lead to new design choices. This paper describes how column-oriented storage techniques can be incorporated in Hadoop in a way that preserves its popular programming APIs. We show that simply using binary storage formats in Hadoop can provide a 3x performance boost over the naive use of text files. We then introduce a column-oriented storage format that is compatible with the replication and scheduling constraints of Hadoop and show that it can speed up MapReduce jobs on real workloads by an order of magnitude. We also show that dealing with complex column types such as arrays, maps, and nested records, which are common in MapReduce jobs, can incur significant CPU overhead. Finally, we introduce a novel skip list column format and lazy record construction strategy that avoids deserializing unwanted records to provide an additional 1.5x performance boost. Experiments on a real intranet crawl are used to show that our column-oriented storage techniques can improve the performance of the map phase in Hadoop by as much as two orders of magnitude.
1. INTRODUCTION
As Hadoop workloads scale, users face performance problems tied to complex types and Hadoop’s programming model. The paper adapts column-oriented storage to Hadoop, reporting substantial gains while preserving its APIs.
1. INTRODUCTION
- The paper adapts parallel-DBMS column-storage techniques to Hadoop, which users also choose for ease of use, low-cost scaling, fault tolerance, and programming flexibility.The introduction frames DBMS and MapReduce as having advantages, with parallel DBMSs holding a performance advantage.Source: §1 · p. 1
- Hadoop users scaling workloads face performance pain, while complex types, hand-coded map and reduce functions, and Java contribute to recurring Hadoop-specific issues.The paper’s motivating example is a bank’s 20-node cluster that could not generate reports in a reasonable time as logs and retention grew.Source: §1 · p. 1
- Join algorithms and indexing are complementary DBMS techniques but lie outside this paper’s scope.The paper focuses on column-oriented storage techniques.Source: §1 · p. 1
1.1 Our Contributions
- The paper designs a column-oriented binary storage format for Hadoop that accounts for HDFS replication and co-locates column data.It assumes a one-time cost to organize input, considered reasonable for datasets analyzed repeatedly.Source: §1.1 · p. 2
- 3x scan-performance improvement comes from switching from text to binary storage formats.The paper notes that text formats had been used in recent Hadoop evaluations, but the benefit of binary formats had not been quantified.Source: §1.1 · p. 2
- Up to 1.5x speedup over eager record construction comes from skip-list columns and lazy record construction.For complex types, lightweight dictionary compression may be preferable to LZO because it incurs lower decompression CPU overhead despite worse compression ratios.Source: §1.1 · p. 2
- The techniques improve Hadoop’s map phase by as much as two orders of magnitude and the overall job by over an order of magnitude.They use Hadoop’s existing extensibility features and require no changes to its core.Source: §1.1 · p. 2
2. HADOOP BACKGROUND
This section introduces Hadoop’s extensibility points for its column-oriented storage format, including how input is divided into map-task splits and converted into typed key-value pairs.
2. HADOOP BACKGROUND
- An InputFormat generates data splits for map tasks and transforms on-disk data into the typed key-value pairs required by the map function.Its methods include specifying input paths, retrieving splits, and obtaining a RecordReader for a split.Source: §2 · p. 2
- SequenceFileInputFormat reads key-value pairs stored in SequenceFiles, a standard serialized binary format.Hadoop also provides InputFormats for text and comma-separated files.Source: §2 · p. 2
- The paper assumes values use Record objects, whose attributes are accessed by name and whose bytes can be handled by serialization frameworks such as Avro, Protocol Buffers, or Thrift.A split is a non-overlapping input partition assigned to a map task.Source: §2 · p. 2
3. CHALLENGES
Complex types add deserialization and compression challenges, while Hadoop’s programming API makes some advanced column-oriented techniques difficult to use in hand-coded jobs.
3.1 Complex Data Types
- Arrays, maps, and nested records create deserialization costs and make effective column-oriented compression difficult.The section illustrates these types with metadata and annotations stored as maps and inlinks stored as an array.Source: §3.1 · p. 3
3.2 Serialization and Deserialization
- Deserialization and object creation can make CPU overhead a Hadoop bottleneck, even when scanning simple types such as integers.The paper reports an experiment measuring this overhead.Source: §3.2 · p. 3
3.3 Compression
- Lightweight compression matters in Hadoop: ZLIB has substantial decompression overhead, while LZO trades compression ratio for lower overhead.The paper describes a lightweight dictionary scheme for complex types as offering better performance than LZO and ZLIB.Source: §3.3 · p. 3
3.4 Query Language vs. Programming API
- Hand-coded MapReduce jobs make techniques such as operating on compressed data, SIMD, and late materialization difficult to use; the paper focuses on Hadoop’s programming API.The described techniques are also applicable to declarative Hadoop languages such as Pig, Hive, and Jaql.Source: §3.4 · p. 3
4. COLUMN-ORIENTED STORAGE
The paper designs a column-oriented format for Hadoop that addresses split parallelism and column co-location, while enabling projection and flexible dataset updates. Its benefits come with a need for a special HDFS placement policy and a possible parallelism limit on smaller datasets.
4.1 Replication and Co-location
- Separate column files require roughly equal splits and co-location; HDFS’s default placement policy provides no co-location guarantees.RCFile packs column-oriented row-groups into HDFS blocks to avoid those challenges, but the paper describes drawbacks in I/O elimination, tuning, and metadata overhead.Source: §4.1 · pp. 3–4
4.2 The CIF Storage Format
- CPP co-locates the files for different columns of a split across replicas, while COF organizes each split-directory as per-column files plus a schema file.CPP relies on a naming convention for split subdirectories; other files use HDFS’s default placement policy.Source: §4.2 · p. 4
- CIF scans selected columns and populates records only with selected fields, leaving the remaining column files unscanned.CIF can assign one or more split-directories to a split and reassembles records from corresponding positions in the column files.Source: §4.2 · p. 4
4.3 Discussion
- Adding a column to CIF requires placing another file in each split-directory, unlike RCFile, which requires rewriting the dataset.This flexibility supports adding columns derived from existing data or moving additional columns from raw inputs into organized storage; the paper says CIF does not pay a performance penalty for this flexibility over RCFile.Source: §4.3 · p. 4
- CIF may limit parallelism on smaller datasets: with c blocks per split-directory, full parallelism requires a dataset exceeding m × c blocks for m map slots.For the paper’s example of 200 map slots and 10 columns, the stated threshold is 128GB; CIF also requires installing a special HDFS block-placement policy.Source: §4.3 · p. 5
5. LAZY RECORD CONSTRUCTION
Lazy record construction postpones deserialization until a map function accesses a field, and skip lists let it skip records efficiently. The section also presents two compression schemes that support lazy decompression.
5. LAZY RECORD CONSTRUCTION
- Lazy record construction deserializes only columns actually accessed by the map function, avoiding metadata deserialization for records whose URLs do not match the filter.The example reads the URL for each record but accesses metadata only when the URL contains “ibm.com/jp”.Source: §5 · p. 5
5.1 Implementation
- LazyRecord tracks the current split record with curPos and each column’s last deserialized record with lastPos; values are read only when a get() method is called.The separate pointers are needed for variable-length complex types, since updating per-column positions otherwise would require deserializing all columns on each next-record call.Source: §5.1 · p. 5, Figure 6 · p. 5
5.2 Skip List Format
- Skip blocks store byte offsets for skipping the next N records, allowing LazyRecord to skip records without deserializing or reading each one individually.The skip-list cost is paid at load time, and the paper reports that the additional loading overhead is minimal.Source: §5.2 · p. 5
5.3 Compression
- Compressed blocks and dictionary-compressed skip lists both support lazy decompression, but dictionary-compressed skip lists allow map values to be accessed without decompressing an entire block.The dictionary scheme targets map-typed columns and can have a worse compression ratio than LZO while using less CPU for decompression.Source: §5.3 · pp. 5–6
6. EXPERIMENTS
The experiments compare CIF with text, SequenceFiles, and RCFile, then test CIF variants on a real intranet crawl. CIF-DCSL delivered the strongest reported results: a 107.8x map-time speedup over SEQ-custom and a 12.8x full-job speedup.
6.1 Experimental Setup
- The experiments used a 40-node Hadoop cluster for HDFS and MapReduce, with six map slots and one reducer per node.Two additional nodes ran the jobtracker and namenode.Source: §6.1 · p. 6
6.2 Benefits of Column-Oriented Storage
- 3x faster scans with SEQ than text and 2.5x–95x faster scans of single columns with CIF than SEQ were reported in the microbenchmark.Scanning all columns with CIF took about 25% longer than with SEQ, attributed to additional seeks across column files.Source: §6.2 · p. 6, Figure 7 · p. 6
- For a single integer projection, CIF was nearly 38x faster than uncompressed RCFile and 33x faster than compressed RCFile.The uncompressed RCFile read 20x more bytes than CIF when scanning one column.Source: §6.2 · pp. 6–7
6.3 Comparison of Column Layouts
- 60.8x better map time than SEQ-custom was achieved by CIF, largely because it read 31.7x less data.CIF-ZLIB read 36GB versus 96GB for CIF but was slightly slower, as decompression CPU overhead outweighed the data reduction.Source: §6.3 · p. 7
- 107.8x map-time speedup over SEQ-custom came from CIF-DCSL, which also gave the best overall results; total job time improved 12.8x over SEQ-custom.Table 1 compares storage formats using speedup ratios relative to SEQ-custom.Source: §6.3 · p. 7, Table 1 · p. 7
6.4 Impact of Co-Location
- 5.1x better map time with CPP than without CPP showed the benefit of co-locating CIF column files.CPP avoided remote access to column files.Source: §6.4 · p. 7
7. RELATED WORK
Related work explores bringing DBMS techniques to MapReduce while retaining its APIs, but this paper focuses on Java MapReduce programs and column-oriented storage. It distinguishes its Hadoop approach from systems such as HadoopDB and Dremel.
7. RELATED WORK
- HadoopDB uses database nodes for execution and MapReduce for scheduling and communication, while later work explored less disruptive indexing and optimization for declarative languages,, [22],.This paper instead focuses on MapReduce programs written directly in Java.Source: §7 · p. 8
- Advanced column-store techniques such as compression integrated with query execution, late materialization, and SIMD are challenging to adapt without a declarative language and specialized runtime,,,.Source: §7 · p. 8
- Unlike Dremel, this work stores complex types in single columns rather than shredding nested data, and also handles map types.The paper focuses on Hadoop and Java, and describes its techniques as complementary to Dremel.Source: §7 · p. 8
8. CONCLUSIONS
The conclusion reports that a Hadoop-compatible column-oriented format improved map-phase performance by as much as two orders of magnitude and full-job time by over one order of magnitude. The appendix describes the generic Record abstraction used by the implementation.
8. CONCLUSIONS
- Column-oriented storage, lightweight compression, and lazy record construction improved map-phase performance by as much as two orders of magnitude and full-job time by over one order of magnitude on an intranet crawl.The format preserves Hadoop’s programming APIs and requires no changes to Hadoop’s core.Source: §8 · p. 8
A. RECORD ABSTRACTION
- The implementation assumes MapReduce jobs use a generic record abstraction based on Avro’s Record interface; LazyRecord and EagerRecord implement it, and ColumnInputFormat supplies Record values.Source: A. RECORD ABSTRACTION · p. 9
- Record attributes are accessed by name through get(name), which returns Object and therefore requires casting to access field values.The Record class supports records conforming to any schema.Source: A. RECORD ABSTRACTION · p. 9
- Avro, Thrift, and Protocol Buffers support complex schema fields, but the paper used Avro; its optional code generation permits generic records without generated schema code.Thrift and Protocol Buffers require code generation by default.Source: A. RECORD ABSTRACTION · p. 9
B.1 Cost of Deserialization
The experiment illustrates that deserialization and object creation can impose substantial CPU overhead, especially for maps.
B.1 Cost of Deserialization
- The test scanned one million 1,000-byte records with varying fractions of integers and byte-array content, using cached data and no disk I/O.The experiment also varied data types, including integers, doubles, and maps.Source: §B.1 · p. 9
- Read bandwidth dropped as the fraction of bytes requiring typed deserialization increased, and Java’s drop was greater than C++’s for the tested cases.C++ read bandwidth was substantially higher for integers and doubles than Java’s.Source: §B.1 · p. 9
- When map values exceeded 60% of a record, their deserialization rate could fall below the bandwidth of a typical SATA disk.Each tested map had four entries with mutable-string keys and integer values; map deserialization required creating new objects.Source: §B.1 · p. 9, Figure 8 · p. 9
B.2 Tuning RCFile
RCFile’s scan performance depends on row-group size: larger groups reduce I/O, but weaken lazy-decompression benefits. CIF avoids this tuning trade-off and performs better in the reported scans.
B.2 Tuning RCFile
- For a scan of a single integer, CIF read 415MB, while RCFile read 16.5GB, 8.5GB, and 4.5 GB with 1MB, 4MB, and 16MB row-group settings, respectively.Source: §B.2 · p. 9
- Larger RCFile row groups improve I/O elimination but harm lazy decompression, while CIF avoids this tuning parameter and is more robust and faster than RCFile.The experiment varied RCFile row-group sizes across 1MB, 4MB, and 16MB.Source: §B.2 · pp. 9–10
- At 16MB row-group size, RCFile’s performance degradation was 2x-3x for single-string and single-map scans.The single-integer scan was RCFile’s worst case.Source: §B.2 · p. 10
B.3 Load Times
The load-time experiment compares conversion from SEQ to several formats. The reported skip-list overhead is fairly minor, while CIF-SL’s savings vary with predicate selectivity and deserialization cost.
B.3 Load Times
- The conversion experiment measured the time to convert the synthetic dataset from SEQ to various formats, with load times presented in Table 2.The selectivity experiment used the Section 6.2 dataset, varied a map-function predicate, and compared CIF with CIF-SL.Source: §B.3 · p. 10, Table 2 · p. 10, Figure 11 · p. 10
- Adding skip lists to CIF had fairly minor overhead, and converting to uncompressed RCFiles took approximately the same time as CIF.The authors attribute skip-list overhead to double buffering under HDFS’s append-only API; the largest skip is limited by main-memory size.Source: §B.3 · p. 10, Table 2 · p. 10
- For highly selective queries, CIF-SL saves more time by eliminating unnecessary deserialization and object creation, then converges to CIF as selectivity approaches 100%.At 100% selectivity, CIF-SL’s overhead relative to CIF is minor; benefits depend on the complex type and its deserialization cost.Source: §B.3 · p. 10, Figure 11 · p. 10, Figure 10 · p. 10
B.5 Varying Record Size
As records widen, CIF outperforms RCFile when scanning a small number of columns. RCFile’s single-column read bandwidth declines with record width, while CIF remains relatively stable.
B.5 Varying Record Size
- CIF performs better than RCFile in all tested cases when projecting a small number of columns.The tests used datasets with 20, 40, and 80 columns, each about 60GB, and projected one column, 10% of columns, or all columns; RCFile used 16MB row groups.Source: §B.5 · p. 11
- As the number of columns increases, CIF’s overhead over SEQ when scanning all columns becomes greater.This pattern is consistent with previous research on column-oriented storage overheads.Source: §B.5 · p. 11
- As records widen, RCFile’s read bandwidth for one column declines, while CIF’s remains relatively stable because each row group covers fewer records.With a 16MB row group, a wider row leaves less data for each column and spreads row-group processing overhead across fewer records.Source: §B.5 · p. 11