Distributed storage: HDFS, file formats and Parquet

1. What HDFS is

The definition the course uses is worth taking apart word by word, because every clause is a design decision that shows up later as a limitation:

Four assumptions follow from that sentence, and they are the ones to quote at the exam:

Read those four together and you get a system that is unapologetically specialised. It will happily stream a terabyte to a hundred machines at once; it will refuse, politely but firmly, to be a database. The fourth assumption is the cultural one: on a commodity cluster (Chapter 2) failure is not an emergency to be handled by an operator, it is a routine event that the filesystem must absorb by itself while the job keeps running.

2. The three goals of a DFS

Before looking at any implementation detail, the deck states what any distributed file system must achieve, and — crucially — the typical solution for each. This table is the architecture of HDFS in miniature; everything in sections 3 to 8 is one of these three rows expanded.

3. Blocks, and why they are so big

Why blocks at all? Two reasons. Files can be larger than disks, so a file must be breakable across machines. And by splitting files into blocks, storage management is simplified and replication is easier: you replicate a fixed-size unit, not a file of arbitrary length.

The block is the minimum amount of data read or written. The scale of that unit is where HDFS parts company with everything you have used before:

An important detail that surprises people: if a file is smaller than a block, it will occupy the necessary disk space, not the full block size. The 128 MB is a maximum unit of splitting, not a minimum unit of allocation.

Why this big?

Because large files split into many small blocks require a huge number of seeks. The slides make it concrete: with a block size of 4 KB, a 1 GB file is 250,000 blocks — and therefore up to 250,000 seeks. On spinning media a seek costs milliseconds while streaming a megabyte costs a fraction of that, so a small block size turns a throughput problem into a latency problem. Big blocks amortise the seek over a large sequential read, which is exactly what the “streaming access” assumption asked for.

4. NameNode and DataNodes

Nodes in an HDFS cluster operate in a master-slave pattern, and the split of responsibilities is unusually clean: one machine knows where everything is, and all the others hold the actual bytes.

Two words in that table carry a lot of weight. Persistently: the tree and the metadata survive a restart, they are written to disk. In memory: the block pool — which block lives on which DataNode — is not persisted, it is rebuilt from the reports the DataNodes send. That is why a NameNode restart is slow (section 5) and why the amount of RAM on the NameNode is the hard ceiling on the number of files in the cluster (sections 6 and 9).

5. The single point of failure and its remedies

The design has an obvious weak spot, and the course names it directly: the NN is a single point of failure. Without it the filesystem cannot be used — there is no way to reconstruct the files from the blocks in the DataNodes. The blocks are still there, intact and replicated, and they are useless: nothing records which block belongs to which file, in which order.

Three remedies, in increasing order of cost and effectiveness:

6. Federation

High availability solves the failure problem of a single master. It does not solve the capacity problem: the size of the block pool is limited by the memory size of the NameNode, which may incur scaling issues on large clusters with many files.

The solution is federation: configure additional NameNodes, where each NN manages a portion of the filesystem (a namespace), and the NNs are independent of each other. The DataNodes are shared; the namespaces are not.

7. Replication and block placement

Each data block is independently replicated at multiple DataNodes in order to improve performance and robustness. Two properties matter: replication is aware of the cluster topology, and for each data block the NN stores the list of DNs storing it.

The default replication factor is 3, and the placement is not random. Memorise the rule — it is a classic exam question:

Replicas can be rebalanced when nodes are added or become unavailable.

Read the three rules as a single sentence and the reasoning appears: one copy where it is free, one copy far enough to survive a rack failure, and the third copy next to the second because the expensive hop has already been paid. Placing replica 3 in a third rack would double the inter-rack traffic to buy protection against the simultaneous loss of two racks — a trade the default configuration refuses. This is Chapter 2 speaking again: intra-rack bandwidth is much greater than inter-rack bandwidth, so the topology of the network dictates the topology of the data.

8. Erasure Coding

Replication is simple and fast, and it is also brutally expensive: the default factor of 3 means 200% overhead — you buy three disks to store one disk of data. HDFS therefore supports Erasure Coding (EC) as an alternative, a mechanism similar to RAID 5-6.

Instead of replicating each block, blocks are striped. A stripe is a sequence of chunks, composed of m data chunks + k parity chunks. Note the consequence the slides call out explicitly: the data in a block may end up in multiple stripes.

EC works best for cold datasets with relatively low I/O activities. The trade-offs, exactly as the slides list them:

9. Where HDFS is not a good fit

The deck is refreshingly honest about the limits. “Although this may change in the future, there are areas where HDFS is not a good fit today.”

10. HDFS I/O: the read path

An application client wishing to read a file (or a portion thereof) must first contact the NN to determine where the actual data is stored. In response, the NN returns:

• the relevant block id;
• the location where the block is held, i.e. which DataNode.

The client then contacts the DataNode to retrieve the data. Blocks are themselves stored on standard single-machine file systems: HDFS lies on top of the standard OS stack — a DataNode is, at bottom, an ordinary Linux box holding ordinary files.

11. Why big data file formats

Storage is only half the story. Data usually comes in standard file formats — structured CSV files, semi-structured JSON files, unstructured textual files — and those formats are, for this setting, actively bad. Three properties separate a big data format from a text file:

Splittability deserves a second look, because it connects directly to everything in Chapter 4. If a file cannot be split, the framework cannot create one map task per block: the whole file has to go to a single task, and the cluster idles. A gzipped CSV is the classic trap — perfectly compressed, and perfectly unparallelisable.

12. Row-oriented vs column-oriented

Everything starts from a physical fact: disks store data sequentially — they are 1-dimensional. Tables cannot be directly represented, so cells are stored either row-by-row or column-by-column. That single choice determines which workload the file is good at.

Row orientation is better for transactional workloads (OLTP), because applications work on a row basis — updating a customer’s profile, adding a new order. Column orientation is better for analytical workloads (OLAP), because analyses are done on a columnar basis — the average salary of employees by department.

The advantages of columnar formats, as listed in the deck:

• Better compression — data is more homogeneous; the space savings are very noticeable at the scale of a Hadoop cluster; type-specific encodings become possible.
• Reduced I/O for analytical queries — only a subset of the columns is read, and unnecessary deserialization is skipped.
• Operate on encoded data — e.g., when a dictionary encoding is used, some operations can run without decoding at all.

The format families

13. Parquet: history and data model

Parquet is a columnar storage format for efficient querying of nested structures in a flat format. Hold on to the tension in that sentence: nested structures, stored flat. Sections 14 and 15 exist entirely to resolve it.

• 2006 — the Dremel system, using the ColumnIO file format, goes into production at Google.
• 2010 — Google publishes a paper about Dremel.
• 2013 — Apache Parquet 1.0 is released (open source) as a joint effort between Twitter and Cloudera.

Parquet ensures interoperability through several frameworks (Spark, MapReduce, etc.) and query engines (Hive, Impala, Pig, Presto, etc.) — which is precisely why it became the default currency of the data lake, and why Chapter 12 will find it again underneath the lakehouse.

The data model

The model is deliberately small — three concepts:

• Root: a group of fields called a message.
• Field types: group or primitive (e.g., string).
• Frequency: required (exactly one), optional (zero or one), or repeated (zero or more).

With only those three frequencies you can express the collection types other systems build in: a list is a repeated field, a map is a repeated group of key-value pairs. The slides show both:

message ExampleList {
  repeated string list;
}

{
  list: "a",
  list: "b",
  list: "c",
  ...
}

message ExampleMap {
  repeated group map {
    required string key;
    optional string value;
  }
}

{
  map: {
    key: "AL",
    value: "Alabama",
  },
  ...
}

14. Unnesting: definition and repetition levels

The problem. Can we store nested data structures in columnar format? We need to map the schema to a list of columns in a way that we can write records to flat columns and read them back to their original nested data structure. And with repeatable fields, how do I know when one record ends and another begins?

The starting rule is simple: one column per primitive-type field in the schema — if we represent the schema as a tree, the primitive types are the leaves of this tree. But a flat column of values loses two things: which nulls happened at which nesting depth, and where each repeated list starts.

Definition level

Consider a column a.b.c where all fields are optional and can be null:

message ExampleDefLevel {
  optional group a {
    optional group b {
      optional string c;
    }
  }
}

When c is defined, then necessarily a and b are defined too. But when c is null, we need to save the level of the null value — is a itself missing, or is a present with b missing, or are both present and only c missing? There are 3 nested optional fields, so the maximum definition level is 3.

Now make b required:

message ExampleDefLevel {
  optional group a {
    required group b {
      optional string c;
    }
  }
}

The maximum definition level is now 2, as b does not need one: a required field cannot be missing on its own, so it contributes no level. Remember: the smaller the definition levels, the less the required bits. Marking a field required is not documentation, it is a storage optimisation.

Repetition level

The presence of repeated fields requires storing when new lists are starting in a column of values. The repetition level indicates at which level we have to create a new list for the current value — it can be seen as a marker of when to start a new list and at which level:

The AddressBook example

Take the schema and the two records from the slides:

message AddressBook {
  required string owner;
  repeated string ownerPhoneNumbers;
  repeated group contacts {
    required string name;
    optional string phoneNumber;
  }
}

AddressBook {
  owner: "Julien Le Dem",
  ownerPhoneNumbers: "555 123 4567",
  ownerPhoneNumbers: "555 666 1337",
  contacts: {
    name: "Dmitriy Ryaboy",
    phoneNumber: "555 987 6543",
  },
  contacts: {
    name: "Chris Aniszczyk"
  }
}
AddressBook {
  owner: "A. Nonymous"
}

For the column contacts.phoneNumber, the definition levels are:

So that column, stripped of everything except its two integers and its value, holds exactly what is needed to rebuild:

AddressBook {
  contacts: {
    phoneNumber: "555 987 6543"
  }
  contacts: { }
}
AddressBook { }

And here is the reconstruction, iterating through the column, exactly as the slides walk it:

15. Two exercises on levels

Both exercises use the same schema. Try each one before opening the solution — this is exactly the kind of question that appears at the exam, and reading the answer is not the same as producing it.

message Person {
  required string name;
  repeated group orders {
    required string item;
    optional int quantity;
  }
}

Exercise 1 — define the columns

Given the following data, define the columns (DL, RL, Value) for item and quantity:

[
  { "name": "Alice",
    "orders": [
      { "item": "Book", "quantity": 1 },
      { "item": "Pen", "quantity": null }
    ]
  },
  { "name": "Bob",
    "orders": null
  },
  { "name": "Carol",
    "orders": [
      { "item": "Notebook", "quantity": 3 }
    ]
  },
  { "name": "Dan",
    "orders": []
  }
]

Exercise 2 — rebuild the messages

Now go the other way. Rebuild the messages starting from the (DL, RL, Value) of item and quantity:

[
  { "orders": [
      { "item": "A", "quantity": null },
      { "item": "B", "quantity": 101 }
  ]},
  { "orders": null },
  { "orders": [
      { "item": "C", "quantity": 102 }
  ]},
  { "orders": null },
  { "orders": [
      { "item": "D", "quantity": 103 },
      { "item": "E", "quantity": null },
      { "item": "F", "quantity": 104 },
      { "item": "G", "quantity": null }
  ]}
]

16. Encodings and the physical file layout

The cost of levels, and how it is paid back

Definition and repetition levels cause overhead: each primitive type corresponds to three columns (the value, its definition level, its repetition level). Two facts keep that overhead small:

• Their values are bound by the depth of the schema, resulting in only a few bits used.
• When all fields are required in a flat schema, levels can be omitted entirely.

Then compression takes care of condensing data efficiently:

The definition of new encodings is supported.

The file format hierarchy

Configuration

Row group size: larger row groups → larger column chunks → larger sequential I/O. Large row groups (512MB–1GB) are recommended, since an entire row group might need to be read and we want it to completely fit on one HDFS block. The optimized read setup named in the deck is: 1GB row groups, 1GB HDFS block size, 1 HDFS block per HDFS file.

Data page size: data pages should be considered indivisible. Smaller data pages → more fine-grained reading (e.g., single row lookup). Larger page sizes → less space overhead (fewer page headers) and less parsing overhead (fewer headers to process).

Test your knowledge