Parquet and ORC

1 The story

One question, one column

The canteen manager asks Riya one question every Monday: how many orders of each item last month? Her job reads the whole 19.4 MB CSV of 5 lakh orders, every order id, every student, every timestamp, only to look at one word per line: the item. Kabir stores the same orders as Parquet and runs the same job. It reads 192,526 bytes. That is 100 times less, for exactly the same answer: chai 149,669.

Riya

Sir, it's the same data. How can it read so little?

Arjun Sir

Because of how the bytes are arranged. A CSV or an Avro file stores one whole order after another, so every item name sits between a student id and a timestamp. Parquet and ORC turn the table sideways: all order ids together, all students together, all items together. A job that needs only items reads only that part of the file.

Kabir

And a column full of the same eight words compresses beautifully. Our item column for 2 lakh orders was 92,796 bytes in ORC. The order id column was 576 bytes.

Arjun Sir

Today we write both formats from MapReduce, look inside them, and measure what projection and predicate pushdown save. These two formats are what nearly every data lake stores today.

2 Definition

Columnar file formats

Formal definition (write this in exams)

A columnar file format stores a table column by column instead of row by row. Apache Parquet divides a file into row groups; each row group holds one column chunk per column, made of pages, and a footer at the end of the file stores the schema and per-chunk metadata such as encodings, sizes and min/max statistics. Apache ORC (Optimized Row Columnar) divides a file into stripes, each with index data (statistics every 10,000 rows by default), column data streams and a stripe footer, followed by a file footer and a postscript. Because values of one type are stored together, both formats apply encodings such as dictionary, run-length (RLE) and delta encoding before general compression (Snappy, ZSTD, ZLIB). Readers use projection (column pruning) to read only requested columns and predicate pushdown to skip row groups or stripes whose statistics show they cannot match a filter.

In simple words

Keep each column together. Then a query reads only the columns it needs, similar values squeeze very small, and a little summary (min and max) of every block lets the reader skip blocks that can't contain what it's looking for.

Words you'll meet in this lesson

Row group (Parquet)A horizontal slice of rows; inside it, one chunk per column.

Column chunkAll values of one column in one row group, stored as pages.

Stripe (ORC)ORC's row group, about 64 MB by default, with its own index.

FooterMetadata at the end of the file: schema, offsets, statistics.

Dictionary encodingStore each distinct value once; the column holds small numbers.

RLERun-length encoding: “chai × 40” instead of 40 copies.

ProjectionReading only the columns a query asks for (column pruning).

Predicate pushdownSkipping row groups whose min/max can't match the filter.

From our lab (Hadoop 3.5.0, 5 lakh orders, written by MapReduce)

Size	CSV 19,418,699 · Avro deflate 4,793,471 · Parquet uncompressed 7,428,043 · Snappy 4,438,970 · ZSTD 2,775,449 · ORC uncompressed 9,238,947 · ZLIB 3,252,876 · ZSTD 2,909,570 bytes
Inside ORC	One part file, 217,602 rows: order_id 576 bytes, item 92,796, student 352,126, time 901,599; student and item dictionary-encoded (2,000 and 8 distinct values)
Projection	Same item-count job: CSV read 19,427,239 bytes; Parquet all columns 4,431,110; Parquet item only 192,526; ORC item only 263,343; every run: chai 149,669
Pushdown	Filter student = 'S1413' on 10 row groups: unsorted file must read 10 of 10; file sorted by student reads 1 of 10. Both return 184 rows

Columnar formats at three levels

Pick your level. New to this? Start at Beginner and climb.

Beginner A register turned sideways

The canteen keeps a bill book: one bill per line with number, student, time, item and amount. That is a row format.
Now imagine five separate registers: one with only bill numbers, one with only students, one with only items… That is a column format.
To count items, you pick up just the item register. And because the item register only contains eight different words, you can write it in shorthand: “chai, chai, chai” becomes “chai ×3”.

Intermediate Where the savings come from

Projection: the reader opens the footer, finds where the item column lives in each row group, and reads only those byte ranges. Our job read 192,526 instead of 4,431,110 bytes from the same Parquet files.
Encoding: in ORC, sequential order ids became 576 bytes for 2 lakh rows (delta + RLE: “start at 1, add 1, 217,602 times”). Items became 3-bit dictionary codes.
Compression on top: similar values sit next to each other, so Snappy, ZSTD or ZLIB work better than on mixed rows. Parquet ZSTD was 7× smaller than the CSV.
Statistics: each row group (Parquet) or stripe and 10,000-row index group (ORC) stores min and max per column, so a filter can skip what can't match.

Advanced Designing for the reader

Pushdown only helps when values are clustered. In our unsorted file every row group held students S0001 to S2000, so nothing could be skipped; sorted by student, one of ten row groups was enough. Sort or partition by the columns people filter on.
Row group or stripe size is a trade-off: bigger means better compression and fewer seeks, smaller means finer skipping and less memory while writing. Parquet's writer default is 128 MB; ORC's stripe default is 64 MB.
Columnar files are written in big batches. They are bad for row-at-a-time appends and updates, and many tiny Parquet files bring back the small files problem (Lesson 7.5).
Table formats (Apache Iceberg, Delta Lake, Apache Hudi) keep Parquet or ORC files as the data and add a transaction log, schema evolution and file-level statistics on top.

Exam tip: For “Why are columnar formats faster for analytics?”, give three reasons: projection (read only needed columns), better compression (same-type values together, dictionary and RLE encodings) and predicate pushdown (min/max statistics per row group or stripe). Name the structures: Parquet file → row groups → column chunks → pages + footer; ORC file → stripes (index, data, stripe footer) + file footer + postscript.

3 Types and ways to use it

Parquet and ORC compared

Feature	Apache Parquet	Apache ORC
Origin	Twitter and Cloudera, 2013 (based on Google's Dremel paper)	Hortonworks for Hive, 2013
Horizontal unit	Row group (writer default 128 MB) → column chunks → pages	Stripe (default 64 MB) with index, data and stripe footer
Statistics	Min/max/null count per column chunk and page; optional bloom filters	Min/max/sum at file, stripe and every 10,000 rows; optional bloom filters
Nested data	Repetition and definition levels (Dremel), very strong	Struct, list and map types
Our lab size	Snappy 4,438,970 · ZSTD 2,775,449 bytes	ZLIB 3,252,876 · ZSTD 2,909,570 bytes
Where it's the default	Spark, Impala, Trino/Presto, Iceberg, Delta, most cloud engines	Hive (required for Hive ACID tables), also read by Spark and Trino

Choosing a layout for the job

Analytics tables

Few columns read, many rows scanned, filters on dates or ids: Parquet or ORC, sorted or partitioned by the filter columns.

columnar

Event streams and exchange

Record-at-a-time writes, whole records read, schemas that change: Avro (Lesson 10.1), then compact to Parquet later.

row

MapReduce intermediates

Key-value data between chained jobs: SequenceFile with block compression (Lesson 10.1).

row

Lakehouse tables

Parquet (or ORC) files managed by Iceberg, Delta Lake or Hudi for transactions, time travel and updates.

columnar + log

Pick the compression codec separately from the format. In our lab Parquet went from 7.4 MB (uncompressed) to 4.4 MB with Snappy and 2.8 MB with ZSTD. Snappy is the common default for speed; ZSTD is now popular for cold data because it is smaller at similar read speed. Lesson 10.3 compares codecs.

4 Step by step

Writing, opening and reading columnar files

On the Module 8 lab with the Module 9 dataset (student_orders.csv). Hadoop ships Avro but not Parquet or ORC, so get_jars.sh downloads them and formats_env.sh puts them on the classpath (see “Run it on your PC”).

Write Parquet and ORC from a MapReduce job

// Parquet: schema + ExampleOutputFormat + codec
GroupWriteSupport.setSchema(schema, job.getConfiguration());
job.setOutputFormatClass(ExampleOutputFormat.class);
ParquetOutputFormat.setCompression(job, CompressionCodecName.SNAPPY);

// ORC: schema string + OrcOutputFormat + codec
OrcConf.MAPRED_OUTPUT_SCHEMA.setString(conf, "struct<order_id:bigint,student:string,time:string,item:string,amount:int>");
OrcConf.COMPRESS.setString(conf, "ZLIB");
job.setOutputFormatClass(OrcOutputFormat.class);

A map-only job: each mapper turns CSV lines into a Parquet Group or an OrcStruct. The output format buffers rows in memory and writes a whole row group or stripe at a time, column by column.

Compare the sizes
```
bash columnar_lab.sh      # writes 6 variants, then hdfs dfs -du -s each
```
text.csv 19418699 avro_deflate 4793471 parquet_none 7428043 parquet_snappy 4438970 parquet_zstd 2775449 orc_none 9238947 orc_zlib 3252876 orc_zstd 2909570
Even without any codec, Parquet is 2.6× smaller than the CSV, only from encodings. With ZSTD it is 7× smaller, and smaller than row-oriented Avro with deflate.
When the first try fails
```
Error: java.lang.ClassNotFoundException: com.github.luben.zstd.RecyclingBufferPool
```
Our first Parquet ZSTD job failed in every map task. Parquet's ZSTD codec needs the zstd-jni jar, which Hadoop keeps in share/hadoop/tools/lib, not on the task classpath. Adding it to -libjars fixed it. ClassNotFoundException in a task almost always means a jar that the client had but the containers didn't.
Open an ORC file
```
java -jar lib/orc-tools-1.9.5-uber.jar meta /tmp/orders.orc
```
Rows: 217602 Compression: ZLIB Type: struct<order_id:bigint,student:string,time:string,item:string,amount:int> Column 1: count: 217602 hasNull: false bytesOnDisk: 576 min: 1 max: 217602 Column 2: count: 217602 hasNull: false bytesOnDisk: 352126 min: S0001 max: S2000 Column 3: count: 217602 hasNull: false bytesOnDisk: 901599 min: 2026-09-01 08:00 max: 2026-09-30 20:59 Column 4: count: 217602 hasNull: false bytesOnDisk: 92796 min: chai max: vada pav Column 5: count: 217602 hasNull: false bytesOnDisk: 105756 min: 15 max: 80 Encoding column 2: DICTIONARY_V2[2000] Encoding column 4: DICTIONARY_V2[8]
The footer knows the schema, row count, and per-column statistics and sizes. 2 lakh sequential order ids took 576 bytes. The timestamp column, with almost every value different, is the biggest: columnar formats shine when values repeat.
Read one column, then all of them
```
hadoop jar mr10.jar ItemCountAny -libjars $LIBJARS -D in.format=parquet -D proj=true  /out/f_parquet_snappy /out/r1
hadoop jar mr10.jar ItemCountAny -libjars $LIBJARS -D in.format=parquet -D proj=false /out/f_parquet_snappy /out/r2
```
text_csv 26917 ms HDFS bytes read 19427239 chai=149669 parquet_all_columns 28564 ms HDFS bytes read 4431110 chai=149669 parquet_item_only 25180 ms HDFS bytes read 192526 chai=149669 orc_all_columns 26269 ms HDFS bytes read 3297403 chai=149669 orc_item_only 25064 ms HDFS bytes read 263343 chai=149669
Projection is one line: parquet.read.schema with only the item field, or orc.include.columns=3. Bytes read fell 23× for Parquet. The run times hardly moved because on 19 MB the 25 s of job start-up dominates; at terabyte scale, bytes read is what decides the time.
Skip row groups with statistics
```
python parquet_inside.py
```
orders.parquet student = 'S1413': must read 10 of 10 row groups [0, 1, 2, 3, 4, 5, 6, 7, 8, 9] -> 184 rows orders_sorted.parquet student = 'S1413': must read 1 of 10 row groups [7] -> 184 rows
Every row group stores min and max per column. In the unsorted file each group spans S0001 to S2000, so the filter can't skip anything. Sorted by student, only row group 7 can contain S1413. Same answer, a tenth of the reading.

Projection and pushdown only work if the reader is told what you need. A job that reads every column and filters in the mapper gets the size benefit but none of the I/O savings. In Spark and Hive this happens automatically from SELECT and WHERE; in plain MapReduce you set the read schema or included columns yourself.

5 See it happen

How a Parquet reader answers “orders of S1413 per item”

Six steps: footer, projection, statistics, skipping, decoding, the answer.

Projection and pushdown 6 steps

What you see

Behind the screen

6 Real-world use

Columnar files at work

1. The data lake's default

Raw events land as JSON or Avro; nightly jobs rewrite them into Parquet tables partitioned by date, which analysts query with Spark, Trino, Hive or a cloud engine.

Raw JSON/Avro→Spark job→Parquet, by date

2. Hive warehouses on ORC

Hive's transactional (ACID) tables, with INSERT, UPDATE and DELETE, must be stored as ORC. Many older Hadoop warehouses keep their big fact tables in ORC with ZLIB or ZSTD.

Hive table→ORC stripes + delta files→Compaction

3. Pay-per-byte query engines

Cloud engines such as Amazon Athena and Google BigQuery on external tables bill by bytes scanned. Moving a table from CSV to compressed Parquet commonly cuts both the bill and the time by an order of magnitude, as our 19.4 MB → 192 KB read shows.

CSV table→Parquet + ZSTD→Fewer bytes scanned

4. Lakehouse tables

Apache Iceberg, Delta Lake and Apache Hudi store data as Parquet files and add a metadata layer with per-file statistics, so engines skip whole files before opening any footer.

Parquet files→Table metadata + stats→File and row-group skipping

Simplified; organisations differ in tools and policies.

7 Run the code

Store the same rows by row and by column

This model stores a few orders row by row (like CSV) and column by column with the two tricks columnar formats use: a dictionary for repeated strings and run-length encoding for repeats in a row. It counts bytes so you can compare.

rows = [(1, "S0001", "chai", 15), (2, "S0001", "chai", 15), (3, "S0002", "chai", 15),
        (4, "S0002", "dosa", 45), (5, "S0003", "chai", 15), (6, "S0003", "chai", 15),
        (7, "S0003", "poha", 25), (8, "S0004", "chai", 15)]

row_bytes = sum(len(",".join(map(str, r))) + 1 for r in rows)          # CSV-style

def dictionary(col):                       # each distinct value once + a small code per row
    d = {v: i for i, v in enumerate(dict.fromkeys(col))}
    return sum(len(str(v)) for v in d) + len(col), [d[v] for v in col]   # 1 byte per code

def rle(codes):                            # (value, run length) pairs
    runs = []
    for c in codes:
        if runs and runs[-1][0] == c: runs[-1][1] += 1
        else: runs.append([c, 1])
    return runs

col_bytes = 0
for name, i in [("order_id", 0), ("student", 1), ("item", 2), ("amount", 3)]:
    col = [r[i] for r in rows]
    if name == "order_id":                  # delta: start + step, like ORC's RLE v2
        size = 2
    else:
        dict_size, codes = dictionary(col)
        size = dict_size - len(col) + 2 * len(rle(codes))
    col_bytes += size
    print(f"{name:9} {size:3} bytes")
print(f"row layout {row_bytes} bytes, column layout {col_bytes} bytes")

order_id 2 bytes student 28 bytes item 22 bytes amount 16 bytes row layout 128 bytes, column layout 68 bytes

def dictionary(col): # each distinct value once + a small code per rowOur real item column has only 8 distinct values: ORC wrote DICTIONARY_V2[8] and stored 2 lakh items in 92,796 bytes.
size = 2Sequential ids are “start at 1, add 1”: ORC stored 217,602 order ids in 576 bytes.
size = dict_size - len(col) + 2 * len(rle(codes))Runs of the same code collapse into one (value, count) pair.

Run it on your PC: write, inspect and read Parquet and ORC

Use the Module 8 lab and the Module 9 dataset (student_orders.csv in HDFS). Hadoop ships Avro but not Parquet or ORC, so first run get_jars.sh inside the namenode container (in /tmp/mr10), then source formats_env.sh before every job: it puts the jars on your client classpath and ships them to the tasks with -libjars. Compile with javac -cp $(hadoop classpath):$HADOOP_CLASSPATH -d classes *.java && jar cf mr10.jar -C classes .

Classpath for the format jars

# formats_env.sh: put the Avro-MapReduce, Parquet and ORC jars on the client and task classpaths
L=/tmp/mr10/lib
export LIBJARS=$(ls $L/*.jar | grep -vE "tools|cli" | paste -sd,)
export HADOOP_CLASSPATH=$(echo $LIBJARS | tr , :)
export LIBJARS=$LIBJARS,$(ls /opt/hadoop/share/hadoop/tools/lib/zstd-jni-*.jar)   # Parquet ZSTD needs zstd-jni (in tools/lib, not on the task classpath)
export HADOOP_CLASSPATH=$(echo $LIBJARS | tr , :)

# In /tmp/mr10 inside the namenode container:
#   bash get_jars.sh
#   source formats_env.sh

(no output: it only sets LIBJARS and HADOOP_CLASSPATH) Without the zstd-jni line, our first Parquet ZSTD job failed in every map task: Error: java.lang.ClassNotFoundException: com.github.luben.zstd.RecyclingBufferPool

Real error from our first run. zstd-jni ships in share/hadoop/tools/lib, which is not on the task classpath by default.

Write Parquet and ORC from MapReduce

// ToColumnar.java: map-only job. CSV orders -> Parquet (-D col.format=parquet) or ORC (-D col.format=orc).
//   -D col.codec=UNCOMPRESSED|SNAPPY|ZSTD|GZIP (Parquet)   NONE|SNAPPY|ZSTD|ZLIB (ORC)
import java.io.IOException;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
import org.apache.hadoop.util.GenericOptionsParser;
import org.apache.orc.OrcConf;
import org.apache.orc.TypeDescription;
import org.apache.orc.mapred.OrcStruct;
import org.apache.orc.mapreduce.OrcOutputFormat;
import org.apache.parquet.example.data.Group;
import org.apache.parquet.example.data.simple.SimpleGroupFactory;
import org.apache.parquet.hadoop.ParquetOutputFormat;
import org.apache.parquet.hadoop.example.ExampleOutputFormat;
import org.apache.parquet.hadoop.example.GroupWriteSupport;
import org.apache.parquet.hadoop.metadata.CompressionCodecName;
import org.apache.parquet.schema.MessageType;
import org.apache.parquet.schema.MessageTypeParser;

public class ToColumnar {
  static final String PARQUET_SCHEMA = "message order { required int64 order_id; required binary student (UTF8); "
      + "required binary time (UTF8); required binary item (UTF8); required int32 amount; }";
  static final String ORC_SCHEMA = "struct<order_id:bigint,student:string,time:string,item:string,amount:int>";

  public static class ParquetMapper extends Mapper<LongWritable, Text, Void, Group> {
    private SimpleGroupFactory factory;
    @Override
    protected void setup(Context ctx) { factory = new SimpleGroupFactory(GroupWriteSupport.getSchema(ctx.getConfiguration())); }
    @Override
    protected void map(LongWritable k, Text line, Context ctx) throws IOException, InterruptedException {
      String[] f = line.toString().split(",");
      Group g = factory.newGroup().append("order_id", Long.parseLong(f[0])).append("student", f[1])
          .append("time", f[2]).append("item", f[3]).append("amount", Integer.parseInt(f[4]));
      ctx.write(null, g);
    }
  }

  public static class OrcMapper extends Mapper<LongWritable, Text, NullWritable, OrcStruct> {
    private final OrcStruct row = (OrcStruct) OrcStruct.createValue(TypeDescription.fromString(ORC_SCHEMA));
    @Override
    protected void map(LongWritable k, Text line, Context ctx) throws IOException, InterruptedException {
      String[] f = line.toString().split(",");
      row.setFieldValue(0, new LongWritable(Long.parseLong(f[0])));
      row.setFieldValue(1, new Text(f[1]));
      row.setFieldValue(2, new Text(f[2]));
      row.setFieldValue(3, new Text(f[3]));
      row.setFieldValue(4, new IntWritable(Integer.parseInt(f[4])));
      ctx.write(NullWritable.get(), row);
    }
  }

  public static void main(String[] args) throws Exception {
    Configuration conf = new Configuration();
    String[] rest = new GenericOptionsParser(conf, args).getRemainingArgs();
    String fmt = conf.get("col.format", "parquet");
    Job job = Job.getInstance(conf, "to " + fmt);
    job.setJarByClass(ToColumnar.class);
    job.setNumReduceTasks(0);
    if (fmt.equals("parquet")) {
      MessageType schema = MessageTypeParser.parseMessageType(PARQUET_SCHEMA);
      GroupWriteSupport.setSchema(schema, job.getConfiguration());
      job.setMapperClass(ParquetMapper.class);
      job.setOutputKeyClass(Void.class);
      job.setOutputValueClass(Group.class);
      job.setOutputFormatClass(ExampleOutputFormat.class);
      ParquetOutputFormat.setCompression(job, CompressionCodecName.valueOf(conf.get("col.codec", "SNAPPY")));
    } else {
      OrcConf.MAPRED_OUTPUT_SCHEMA.setString(job.getConfiguration(), ORC_SCHEMA);
      OrcConf.COMPRESS.setString(job.getConfiguration(), conf.get("col.codec", "ZLIB"));
      job.setMapperClass(OrcMapper.class);
      job.setOutputKeyClass(NullWritable.class);
      job.setOutputValueClass(OrcStruct.class);
      job.setOutputFormatClass(OrcOutputFormat.class);
    }
    FileInputFormat.addInputPath(job, new Path(rest[0]));
    FileOutputFormat.setOutputPath(job, new Path(rest[1]));
    System.exit(job.waitForCompletion(true) ? 0 : 1);
  }
}

// Inside the container, after source formats_env.sh:
//   hadoop jar mr10.jar ToColumnar -libjars $LIBJARS -D col.format=parquet -D col.codec=SNAPPY \
//       /user/hadoop/data/student_orders.csv /out/f_parquet_snappy
//   hadoop jar mr10.jar ToColumnar -libjars $LIBJARS -D col.format=orc -D col.codec=ZLIB \
//       /user/hadoop/data/student_orders.csv /out/f_orc_zlib

== size on HDFS (one replica) text.csv 19418699 avro_deflate 4793471 parquet_none 7428043 parquet_snappy 4438970 parquet_zstd 2775449 orc_none 9238947 orc_zlib 3252876 orc_zstd 2909570 == inside one ORC file (orc-tools meta) Rows: 217602 Compression: ZLIB Type: struct<order_id:bigint,student:string,time:string,item:string,amount:int> File Statistics: Column 0: count: 217602 hasNull: false Column 1: count: 217602 hasNull: false bytesOnDisk: 576 min: 1 max: 217602 sum: 23675424003 Column 2: count: 217602 hasNull: false bytesOnDisk: 352126 min: S0001 max: S2000 sum: 1088010 Column 3: count: 217602 hasNull: false bytesOnDisk: 901599 min: 2026-09-01 08:00 max: 2026-09-30 20:59 sum: 3481632 Column 4: count: 217602 hasNull: false bytesOnDisk: 92796 min: chai max: vada pav sum: 1101261 Column 5: count: 217602 hasNull: false bytesOnDisk: 105756 min: 15 max: 80 sum: 5598914 Stripe: offset: 3 data: 1452853 rows: 217602 tail: 124 index: 1637 Encoding column 0: DIRECT Encoding column 1: DIRECT_V2 Encoding column 2: DICTIONARY_V2[2000] Encoding column 3: DIRECT_V2 Encoding column 4: DICTIONARY_V2[8] Encoding column 5: DIRECT_V2

Real output of columnar_lab.sh: the sizes of all six variants (plus the CSV and Avro from Lesson 10.1), then orc-tools meta on one ORC part file.

Count items from any format, with projection

// ItemCountAny.java: orders per item, reading any of our formats.  -D in.format=text|seq|avro|parquet|orc
// For parquet and orc only the "item" column is requested (column projection); -D proj=false reads all columns.
import java.io.IOException;
import org.apache.avro.Schema;
import org.apache.avro.generic.GenericRecord;
import org.apache.avro.mapred.AvroKey;
import org.apache.avro.mapreduce.AvroJob;
import org.apache.avro.mapreduce.AvroKeyInputFormat;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.input.SequenceFileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
import org.apache.hadoop.mapreduce.lib.reduce.LongSumReducer;
import org.apache.hadoop.util.GenericOptionsParser;
import org.apache.orc.OrcConf;
import org.apache.orc.mapred.OrcStruct;
import org.apache.orc.mapreduce.OrcInputFormat;
import org.apache.parquet.example.data.Group;
import org.apache.parquet.hadoop.example.ExampleInputFormat;

public class ItemCountAny {
  static final LongWritable ONE = new LongWritable(1);

  public static class TextM extends Mapper<LongWritable, Text, Text, LongWritable> {
    private final Text item = new Text();
    @Override protected void map(LongWritable k, Text v, Context c) throws IOException, InterruptedException {
      item.set(v.toString().split(",")[3]); c.write(item, ONE);
    }
  }
  public static class SeqM extends Mapper<LongWritable, Text, Text, LongWritable> {   // value = student,time,item,amount
    private final Text item = new Text();
    @Override protected void map(LongWritable k, Text v, Context c) throws IOException, InterruptedException {
      item.set(v.toString().split(",")[2]); c.write(item, ONE);
    }
  }
  public static class AvroM extends Mapper<AvroKey<GenericRecord>, NullWritable, Text, LongWritable> {
    private final Text item = new Text();
    @Override protected void map(AvroKey<GenericRecord> k, NullWritable v, Context c) throws IOException, InterruptedException {
      item.set(k.datum().get("item").toString()); c.write(item, ONE);
    }
  }
  public static class ParquetM extends Mapper<Void, Group, Text, LongWritable> {
    private final Text item = new Text();
    @Override protected void map(Void k, Group g, Context c) throws IOException, InterruptedException {
      item.set(g.getString("item", 0)); c.write(item, ONE);
    }
  }
  public static class OrcM extends Mapper<NullWritable, OrcStruct, Text, LongWritable> {
    @Override protected void map(NullWritable k, OrcStruct row, Context c) throws IOException, InterruptedException {
      c.write((Text) row.getFieldValue("item"), ONE);
    }
  }

  public static void main(String[] args) throws Exception {
    Configuration conf = new Configuration();
    String[] rest = new GenericOptionsParser(conf, args).getRemainingArgs();
    String fmt = conf.get("in.format", "text");
    Job job = Job.getInstance(conf, "item count from " + fmt);
    job.setJarByClass(ItemCountAny.class);
    switch (fmt) {
      case "seq": job.setInputFormatClass(SequenceFileInputFormat.class); job.setMapperClass(SeqM.class); break;
      case "avro":
        job.setInputFormatClass(AvroKeyInputFormat.class);
        AvroJob.setInputKeySchema(job, new Schema.Parser().parse(ToAvro.SCHEMA));
        job.setMapperClass(AvroM.class); break;
      case "parquet":
        job.setInputFormatClass(ExampleInputFormat.class);
        if (conf.getBoolean("proj", true))
          job.getConfiguration().set("parquet.read.schema", "message order { required binary item (UTF8); }");   // projection
        job.setMapperClass(ParquetM.class); break;
      case "orc":
        job.setInputFormatClass(OrcInputFormat.class);
        if (conf.getBoolean("proj", true))
          OrcConf.INCLUDE_COLUMNS.setString(job.getConfiguration(), "3");                                       // only column 3: item
        job.setMapperClass(OrcM.class); break;
      default: job.setMapperClass(TextM.class);
    }
    job.setCombinerClass(LongSumReducer.class);
    job.setReducerClass(LongSumReducer.class);
    job.setOutputKeyClass(Text.class);
    job.setOutputValueClass(LongWritable.class);
    FileInputFormat.addInputPath(job, new Path(rest[0]));
    FileOutputFormat.setOutputPath(job, new Path(rest[1]));
    System.exit(job.waitForCompletion(true) ? 0 : 1);
  }
}

// -D in.format=parquet|orc and -D proj=true|false, e.g.:
//   hadoop jar mr10.jar ItemCountAny -libjars $LIBJARS -D in.format=parquet -D proj=true /out/f_parquet_snappy /out/r_p1

== read: all columns vs only the item column text_csv 26917 ms HDFS bytes read 19427239 chai=149669 parquet_all_columns 28564 ms HDFS bytes read 4431110 chai=149669 parquet_item_only 25180 ms HDFS bytes read 192526 chai=149669 orc_all_columns 26269 ms HDFS bytes read 3297403 chai=149669 orc_item_only 25064 ms HDFS bytes read 263343 chai=149669

Real output. Compare “HDFS bytes read”: projection cut Parquet from 4,431,110 to 192,526 bytes. Times barely differ because 19 MB is tiny next to the ~25 s of job start-up.

The whole lab

# columnar_lab.sh: the same orders as Parquet and ORC: size, what's inside, and reading only one column
cd /tmp/mr10; source formats_env.sh
D=/user/hadoop/data/student_orders.csv
conv() { name=$1; shift
  hdfs dfs -rm -r -f -skipTrash /out/f_$name > /dev/null 2>&1
  hadoop jar mr10.jar ToColumnar -libjars $LIBJARS "$@" $D /out/f_$name > /tmp/conv_$name.log 2>&1 || echo "$name FAILED"; }
conv parquet_none   -D col.format=parquet -D col.codec=UNCOMPRESSED
conv parquet_snappy -D col.format=parquet -D col.codec=SNAPPY
conv parquet_zstd   -D col.format=parquet -D col.codec=ZSTD
conv orc_none   -D col.format=orc -D col.codec=NONE
conv orc_zlib   -D col.format=orc -D col.codec=ZLIB
conv orc_zstd   -D col.format=orc -D col.codec=ZSTD
echo "== size on HDFS (one replica)"
printf "%-15s %12s\n" text.csv $(hdfs dfs -du -s $D | awk '{print $1}')
printf "%-15s %12s\n" avro_deflate $(hdfs dfs -du -s /out/f_avro_deflate | awk '{print $1}')
for n in parquet_none parquet_snappy parquet_zstd orc_none orc_zlib orc_zstd; do
  printf "%-15s %12s\n" $n $(hdfs dfs -du -s /out/f_$n | awk '{print $1}')
done
echo "== inside one ORC file (orc-tools meta)"
f=$(hdfs dfs -ls /out/f_orc_zlib | grep part | head -1 | awk '{print $8}')
hdfs dfs -get -f $f /tmp/orders.orc 2>/dev/null
java -jar lib/orc-tools-1.9.5-uber.jar meta /tmp/orders.orc 2>/dev/null > /tmp/orc_meta.txt
grep -E "^Rows:|^Compression:|^Type:" /tmp/orc_meta.txt
sed -n '/^File Statistics:/,/^$/p' /tmp/orc_meta.txt | cut -c1-150
grep -m1 "Stripe: offset" /tmp/orc_meta.txt; grep -m6 "Encoding column" /tmp/orc_meta.txt
echo "== read: all columns vs only the item column"
read_job() {  # name, input, format, projection
  hdfs dfs -rm -r -f -skipTrash /out/r_$1 > /dev/null 2>&1
  start=$(date +%s%N)
  hadoop jar mr10.jar ItemCountAny -libjars $LIBJARS -D in.format=$3 -D proj=$4 $2 /out/r_$1 > /tmp/read_$1.log 2>&1
  printf "%-22s %6d ms  HDFS bytes read %10s  chai=%s\n" $1 $(( ($(date +%s%N) - start) / 1000000 )) \
    $(grep "HDFS: Number of bytes read=" /tmp/read_$1.log | head -1 | cut -d= -f2) $(hdfs dfs -cat /out/r_$1/part-r-00000 | grep chai | cut -f2)
}
read_job text_csv $D text true
read_job parquet_all_columns /out/f_parquet_snappy parquet false
read_job parquet_item_only /out/f_parquet_snappy parquet true
read_job orc_all_columns /out/f_orc_zlib orc false
read_job orc_item_only /out/f_orc_zlib orc true

# Run inside the namenode container, in /tmp/mr10:  bash columnar_lab.sh

== size on HDFS (one replica) text.csv 19418699 avro_deflate 4793471 parquet_none 7428043 parquet_snappy 4438970 parquet_zstd 2775449 orc_none 9238947 orc_zlib 3252876 orc_zstd 2909570 == inside one ORC file (orc-tools meta) Rows: 217602 Compression: ZLIB Type: struct<order_id:bigint,student:string,time:string,item:string,amount:int> File Statistics: Column 0: count: 217602 hasNull: false Column 1: count: 217602 hasNull: false bytesOnDisk: 576 min: 1 max: 217602 sum: 23675424003 Column 2: count: 217602 hasNull: false bytesOnDisk: 352126 min: S0001 max: S2000 sum: 1088010 Column 3: count: 217602 hasNull: false bytesOnDisk: 901599 min: 2026-09-01 08:00 max: 2026-09-30 20:59 sum: 3481632 Column 4: count: 217602 hasNull: false bytesOnDisk: 92796 min: chai max: vada pav sum: 1101261 Column 5: count: 217602 hasNull: false bytesOnDisk: 105756 min: 15 max: 80 sum: 5598914 Stripe: offset: 3 data: 1452853 rows: 217602 tail: 124 index: 1637 Encoding column 0: DIRECT Encoding column 1: DIRECT_V2 Encoding column 2: DICTIONARY_V2[2000] Encoding column 3: DIRECT_V2 Encoding column 4: DICTIONARY_V2[8] Encoding column 5: DIRECT_V2 == read: all columns vs only the item column text_csv 26917 ms HDFS bytes read 19427239 chai=149669 parquet_all_columns 28564 ms HDFS bytes read 4431110 chai=149669 parquet_item_only 25180 ms HDFS bytes read 192526 chai=149669 orc_all_columns 26269 ms HDFS bytes read 3297403 chai=149669 orc_item_only 25064 ms HDFS bytes read 263343 chai=149669

Real output, about 5 minutes. Run Lesson 10.1's rowformats_lab.sh first if you want the Avro size line filled in.

Inside a Parquet file, and row-group skipping

# parquet_inside.py: look inside a Parquet file and see row-group skipping (predicate pushdown).
# Needs: pip install pyarrow.  Run next to student_orders.csv (from Module 9's make_student_orders.py).
import pyarrow as pa, pyarrow.csv as csv, pyarrow.parquet as pq, pyarrow.compute as pc

names = ["order_id", "student", "time", "item", "amount"]
t = csv.read_csv("student_orders.csv", read_options=csv.ReadOptions(column_names=names))
pq.write_table(t, "orders.parquet", row_group_size=50_000, compression="snappy")
pq.write_table(t.sort_by("student"), "orders_sorted.parquet", row_group_size=50_000, compression="snappy")

md = pq.ParquetFile("orders.parquet").metadata
print(f"rows {md.num_rows:,}  row groups {md.num_row_groups}  created by: {md.created_by}")
rg = md.row_group(0)
print("row group 0, per column chunk:")
for i in range(rg.num_columns):
    c = rg.column(i)
    print(f"  {c.path_in_schema:9} {c.compression:7} {c.total_compressed_size:>8,} bytes  "
          f"encodings {','.join(e for e in c.encodings)}  min={c.statistics.min} max={c.statistics.max}")

def groups_to_read(path, student):
    f = pq.ParquetFile(path)
    hits = []
    for i in range(f.metadata.num_row_groups):
        s = f.metadata.row_group(i).column(1).statistics          # column 1 = student
        if s.min <= student <= s.max:                              # the footer says it MIGHT be here
            hits.append(i)
    return hits, f.metadata.num_row_groups

for path in ["orders.parquet", "orders_sorted.parquet"]:
    hits, total = groups_to_read(path, "S1413")
    rows = pq.read_table(path, filters=[("student", "=", "S1413")]).num_rows
    print(f"{path:22} student = 'S1413': must read {len(hits)} of {total} row groups {hits}  -> {rows} rows")

# On your PC, next to student_orders.csv:
#   pip install pyarrow
#   python parquet_inside.py

rows 500,000 row groups 10 created by: parquet-cpp-arrow version 26.0.0 row group 0, per column chunk: order_id SNAPPY 297,991 bytes encodings PLAIN,RLE,RLE_DICTIONARY min=1 max=50000 student SNAPPY 77,573 bytes encodings PLAIN,RLE,RLE_DICTIONARY min=S0001 max=S2000 time SNAPPY 210,234 bytes encodings PLAIN,RLE,RLE_DICTIONARY min=2026-09-01 08:00:00 max=2026-09-30 20:59:00 item SNAPPY 19,107 bytes encodings PLAIN,RLE,RLE_DICTIONARY min=chai max=vada pav amount SNAPPY 19,147 bytes encodings PLAIN,RLE,RLE_DICTIONARY min=15 max=80 orders.parquet student = 'S1413': must read 10 of 10 row groups [0, 1, 2, 3, 4, 5, 6, 7, 8, 9] -> 184 rows orders_sorted.parquet student = 'S1413': must read 1 of 10 row groups [7] -> 184 rows

Real output with pyarrow 26. pyarrow guessed the time column as a timestamp, so its statistics print with seconds. The unsorted file must read every row group; sorted by student, only row group 7.

8 Line by line

Run-length encoding a column

Columnar files store runs of equal values as (value, count). Follow how a sorted item column collapses.

Worked examples: real problems, solved and traced in memory

Three columnar-format problems: how many bytes a query reads, how sorting changes skipping, and what a format switch saves.

Example 1BeginnerProjection · bytes

How much does a two-column query read?

Problem statement

In one ORC part file, the column sizes on disk are order_id 576, student 352,126, time 901,599, item 92,796 and amount 105,756 bytes. A query needs item and amount. How many column bytes does it read, and what share of all columns is that?

Input

5 column sizes from orc-tools meta

Expected output

read 198,552 of 1,452,853 bytes = 13.7%

Approach

Add up all column sizes.
Add up only the columns the query needs.
Divide.

Solution: run it live

What happens in memory, step by step

Recorded from a real Python run: press Next to execute one line at a time. Watch variables change in each frame, and lists and dictionaries (numbered #1, #2 …) fill up as objects.

Example 2IntermediatePushdown · sorting

Why sorting decides how much you skip

Problem statement

A 10-row-group file holds orders of 2,000 students. Unsorted, every row group contains students S0001 to S2000. Sorted by student, each row group covers a narrow range. For a query on one student, how many row groups must be read in each case, and how many rows is that?

Input

10 row groups of 50,000 rows

Expected output

unsorted: 10 row groups, 500,000 rows decoded for 184 matches
sorted  : 1 row groups, 50,000 rows decoded for 184 matches

Approach

A row group is read if min ≤ value ≤ max.
Unsorted: every range is S0001–S2000.
Sorted: only the group whose range contains the student.

Solution: run it live

What happens in memory, step by step

Recorded from a real Python run: press Next to execute one line at a time. Watch variables change in each frame, and lists and dictionaries (numbered #1, #2 …) fill up as objects.

Example 3AdvancedCost · migration

Is converting a table to Parquet worth it?

Problem statement

A 50 TB CSV table is scanned by 400 queries a month, each reading 3 of 20 columns on average. A query engine charges ₹400 per TB scanned (illustrative). Parquet with ZSTD makes the table 7× smaller and projection reads about 3/20 of it. Converting costs one 50 TB scan. Compare the monthly bill and the payback time.

Input

50 TB · 400 queries · 3 of 20 columns · 7× compression

Expected output

CSV Rs 8,000,000/month, Parquet Rs 171,429/month
conversion Rs 20,000, paid back in 0.1 days

Approach

CSV cost per query = 50 TB × ₹400.
Parquet per query = 50 / 7 × 3/20 TB × ₹400.
Payback = conversion cost / monthly saving.

Solution: run it live

What happens in memory, step by step

Recorded from a real Python run: press Next to execute one line at a time. Watch variables change in each frame, and lists and dictionaries (numbered #1, #2 …) fill up as objects.

9 Quiz and revision

Check yourself

Quick revision

Lesson 10.2 in one screen

Columnar = values of one column stored together: projection, better encoding/compression, predicate pushdown.
Parquet: file → row groups (128 MB default) → column chunks → pages; footer with schema and min/max; magic “PAR1”.
ORC: stripes (64 MB) with index (every 10,000 rows), data streams, stripe footer; file footer + postscript.
Lab sizes: CSV 19.4 MB · Parquet Snappy 4.4 · ZSTD 2.8 · ORC ZLIB 3.3 · ZSTD 2.9 MB.
Lab reads: CSV 19,427,239 bytes · Parquet all columns 4,431,110 · item only 192,526 · ORC item only 263,343.
Pushdown needs clustering: S1413 → 10 of 10 row groups unsorted, 1 of 10 sorted.
Parquet: default in Spark/Trino/Impala/Iceberg/Delta. ORC: Hive, required for Hive ACID.

Interview and viva questions

What is the difference between Parquet and ORC?

Both are columnar formats with encodings, compression and statistics. Parquet organises a file into row groups, column chunks and pages, has strong support for nested data and is the default for Spark, Trino, Impala and table formats like Iceberg and Delta. ORC organises a file into stripes with lightweight indexes every 10,000 rows, came from the Hive project and is required for Hive ACID tables. Sizes and speed are usually similar; ecosystem decides.

Why is Parquet faster than CSV?

A Parquet reader reads only the columns a query needs (projection), the data is much smaller because each column is encoded (dictionary, run-length, delta) and compressed, and min/max statistics let it skip row groups that cannot match a filter. In our lab a job needing one column read 192,526 bytes from Parquet against 19,427,239 bytes from the CSV.

What is predicate pushdown?

Pushing a filter (a predicate such as student = 'S1413') down to the file reader so it can skip data using the statistics stored in the file. If a row group's min and max for a column cannot satisfy the filter, the reader skips the whole row group. It works best when data is sorted or clustered by the filtered column.

What is a row group in Parquet?

A horizontal partition of a Parquet file containing a set of rows, stored as one column chunk per column. Each column chunk is divided into pages. The writer default size is 128 MB; statistics are kept per column chunk (and per page), so the row group is the main unit for skipping and for parallel reading.

Should I use Avro or Parquet?

Use Avro (row-oriented) for writing records one at a time, messaging and data exchange where whole records are read and schemas evolve. Use Parquet (columnar) for analytical tables that are scanned with few columns and filters. A common pipeline writes Avro or JSON on ingest and compacts it into Parquet tables.

Practice problems

Columnar format exercises using the real sizes and reads from our lab.

P1Easy

How much smaller?

Compute how many times smaller each Parquet variant is than the CSV.

Input

19,418,699 vs 7,428,043 / 4,438,970 / 2,775,449

Expected output

parquet uncompressed 2.6x smaller
parquet snappy       4.4x smaller
parquet zstd         7.0x smaller

Show solution and explanation

csv = 19418699
for name, b in [("uncompressed", 7428043), ("snappy", 4438970), ("zstd", 2775449)]:
    print(f"parquet {name:12} {csv / b:.1f}x smaller")

parquet uncompressed 2.6x smaller parquet snappy 4.4x smaller parquet zstd 7.0x smaller

How it works: 2.6×, 4.4× and 7.0×. Even with no codec, encodings alone made Parquet 2.6× smaller.

P2Easy

Projection saving

How many times fewer bytes did the item-only read take than the all-columns read, for Parquet and for ORC?

Input

4,431,110 / 192,526 · 3,297,403 / 263,343

Expected output

parquet 23x, orc 12.5x

Show solution and explanation

print(f"parquet {4431110 / 192526:.0f}x, orc {3297403 / 263343:.1f}x")

parquet 23x, orc 12.5x

How it works: 23× and 12.5×. The item column is one of the cheapest columns, so the saving is large.

P3Easy

Bytes per value

How many bits per row did ORC use for order_id and for item?

Input

576 and 92,796 bytes for 217,602 rows

Expected output

order_id 0.02 bits per row
item     3.41 bits per row

Show solution and explanation

rows = 217602
for name, b in [("order_id", 576), ("item", 92796)]:
    print(f"{name:8} {b * 8 / rows:.2f} bits per row")

order_id 0.02 bits per row item 3.41 bits per row

How it works: 0.02 and 3.4 bits per row. A CSV spends about 6 bytes (48 bits) per order id.

P4Medium

Dictionary encode a column

Dictionary-encode the item column: print the dictionary and the codes.

Input

6 values

Expected output

{'dosa': 0, 'samosa': 1, 'chai': 2}
[0, 1, 1, 2, 0, 2]

Show solution and explanation

col = ["dosa", "samosa", "samosa", "chai", "dosa", "chai"]
d = {}
codes = [d.setdefault(v, len(d)) for v in col]
print(d)
print(codes)

{'dosa': 0, 'samosa': 1, 'chai': 2} [0, 1, 1, 2, 0, 2]

How it works: Three distinct strings stored once; the column becomes small integers that pack into a few bits each.

P5Medium

Delta encoding

Delta-encode a sorted id column and decode it back.

Input

ids 1001..1006

Expected output

encoded: [1001, 1, 1, 1, 1, 1]
decoded: [1001, 1002, 1003, 1004, 1005, 1006] True

Show solution and explanation

ids = [1001, 1002, 1003, 1004, 1005, 1006]
deltas = [ids[0]] + [b - a for a, b in zip(ids, ids[1:])]
print("encoded:", deltas)
back = [deltas[0]]
for d in deltas[1:]:
    back.append(back[-1] + d)
print("decoded:", back, back == ids)

encoded: [1001, 1, 1, 1, 1, 1] decoded: [1001, 1002, 1003, 1004, 1005, 1006] True

How it works: All deltas are 1, which run-length encoding then turns into “1 × 5”: why 2 lakh ids took 576 bytes.

P6Medium

Row groups to read

Given row-group min/max dates, which groups must be read for orders on 2026-09-17?

Input

4 row groups

Expected output

[2]

Show solution and explanation

groups = [("2026-09-01", "2026-09-08"), ("2026-09-08", "2026-09-15"), ("2026-09-15", "2026-09-23"), ("2026-09-23", "2026-09-30")]
print([i for i, (lo, hi) in enumerate(groups) if lo <= "2026-09-17" <= hi])

[2]

How it works: Only group 2. ISO dates compare correctly as strings; that is one reason to store dates in sortable form.

P7Business problem

Storage bill

The canteen chain keeps 3 years of orders: 200 GB of CSV per year, stored with replication 3 at ₹2 per GB per month. What is the monthly storage bill as CSV and as Parquet ZSTD (7× smaller)?

Input

600 GB · ×3 replicas · ₹2/GB-month

Expected output

CSV          Rs 3,600 per month
Parquet ZSTD Rs 514 per month

Show solution and explanation

csv_gb = 3 * 200
for name, gb in [("CSV", csv_gb), ("Parquet ZSTD", csv_gb / 7)]:
    print(f"{name:12} Rs {gb * 3 * 2:,.0f} per month")

CSV Rs 3,600 per month Parquet ZSTD Rs 514 per month

How it works: ₹3,600 vs about ₹514 per month (illustrative prices), before counting the faster queries.

P8Advanced

Choose a sort column

Queries filter by student 70% of the time and by item 30%. With 10 row groups, how many row groups does an average query read if the file is sorted by student, and if sorted by item? Assume a sorted column narrows to 1 group and an unsorted one needs all 10, except that 8 items sorted means about 2 groups per item.

Input

70% student, 30% item

Expected output

sorted by student: 3.7 groups per query, sorted by item: 7.6

Show solution and explanation

by_student = 0.7 * 1 + 0.3 * 10
by_item = 0.7 * 10 + 0.3 * 2
print(f"sorted by student: {by_student:.1f} groups per query, sorted by item: {by_item:.1f}")

sorted by student: 3.7 groups per query, sorted by item: 7.6

How it works: 3.7 vs 7.6 groups: sort by the column most queries filter on. Engines also offer Z-ordering or bloom filters to help a second column.

Real-world scenario questions (asked in company interviews)

File formats come up in every data engineering interview, often as “Parquet or ORC?” or “why is this query slow?”.

Formats · basicsFresher

“Why is Parquet better than CSV for analytics?”

The standard opener.

The interviewer is checking: the three mechanisms, not just “it's compressed”.

Show a strong answer

Columnar: a query reads only the columns it needs (projection).
Each column is encoded (dictionary, RLE, delta) and compressed; similar values compress far better.
Footer statistics (min/max per row group) let readers skip data for filters; the schema travels with the file.

Performance · pushdownExperienced

“Our Parquet table is filtered by customer_id but every query scans the whole table.”

The table is partitioned by date only.

The interviewer is checking: clustering and statistics.

Show a strong answer

Min/max statistics only skip row groups when values are clustered; random order makes every row group span all customers.
Sort (or Z-order) data by customer_id within each date partition when writing or compacting; consider bloom filters for point lookups.
Check the engine's “row groups skipped / bytes scanned” metrics before and after.

Design · choiceExperienced

“Parquet or ORC for our new lake?”

The team uses Spark for ETL and Trino and Hive for queries.

The interviewer is checking: ecosystem reasoning.

Show a strong answer

Both perform similarly; choose by ecosystem. Spark, Trino, Iceberg and Delta default to Parquet.
Pick ORC if you need Hive ACID tables or have an existing ORC warehouse.
Standardise the codec (Snappy for hot, ZSTD for cold) and target file sizes of hundreds of MB.

Operations · failureExperienced

A job writing Parquet with ZSTD fails with ClassNotFoundException in every map task

The same code worked with Snappy, and the class compiles fine on the edge node.

The interviewer is checking: client vs task classpath.

Show a strong answer

The task containers lack a jar the client has (here zstd-jni, which lives in share/hadoop/tools/lib).
Ship it with -libjars (or build a fat jar, or set mapreduce.application.classpath on the cluster).
Read the attempt diagnostics in the JobHistory server to see the exact missing class.

One question, one column

Columnar file formats

Words you'll meet in this lesson

From our lab (Hadoop 3.5.0, 5 lakh orders, written by MapReduce)

Columnar formats at three levels

Beginner A register turned sideways

Intermediate Where the savings come from

Advanced Designing for the reader

Parquet and ORC compared

Choosing a layout for the job

Analytics tables

Event streams and exchange

MapReduce intermediates

Lakehouse tables

Writing, opening and reading columnar files

Write Parquet and ORC from a MapReduce job

Compare the sizes

When the first try fails

Open an ORC file

Read one column, then all of them

Skip row groups with statistics

How a Parquet reader answers “orders of S1413 per item”

Columnar files at work

1. The data lake's default

2. Hive warehouses on ORC

3. Pay-per-byte query engines

4. Lakehouse tables

Store the same rows by row and by column

Run it on your PC: write, inspect and read Parquet and ORC

Classpath for the format jars

Write Parquet and ORC from MapReduce

Count items from any format, with projection

The whole lab

Inside a Parquet file, and row-group skipping

Run-length encoding a column

Worked examples: real problems, solved and traced in memory

How much does a two-column query read?

Why sorting decides how much you skip

Is converting a table to Parquet worth it?

Check yourself

Quick revision

Lesson 10.2 in one screen

Interview and viva questions

Practice problems

How much smaller?

Projection saving

Bytes per value

Dictionary encode a column

Delta encoding

Row groups to read

Storage bill

Choose a sort column

Real-world scenario questions (asked in company interviews)

“Why is Parquet better than CSV for analytics?”

“Our Parquet table is filtered by customer_id but every query scans the whole table.”

“Parquet or ORC for our new lake?”

A job writing Parquet with ZSTD fails with ClassNotFoundException in every map task