Spark SQL, Catalyst and the columnar file formats

Spark SQL is about more than SQL

The title slide says it in five words: Spark SQL "is about more than SQL" (p. 1262). Page 1263 gives the reasons. Big-data work must "perform ETL to and from various (semi- or unstructured) data sources" and "perform advanced analytics (e.g. machine learning, graph processing) that are hard to express in relational systems". The answer: "a DataFrame API that can perform relational operations on both external data sources and Spark's built-in RDDs", plus "a highly extensible optimizer, Catalyst".

Page 1264 turns this into three promises of declarative big-data processing: write less code, read less data, let the optimizer do the hard work. You state what you want; Spark decides how. This lesson follows the data: your code, the optimizer, the file.

Write less code: one average, four dialects

Page 1265 computes an average in about thirty lines of Hadoop Java and five of Spark; p. 1266 writes the average age per name in four dialects: RDDs, SQL, Pig and the DataFrame. Take people(name, age): Ann 30, Bob 20, Ann 40, Cid 10. (The slides' sqlCtx is today's session object spark.)

from pyspark.sql import functions as F
# one row per name: Ann 35.0, Bob 20.0, Cid 10.0
spark.table("people").groupBy("name").agg(F.avg("age")).show()

The RDD version of p. 1266 carries the average's ingredients by hand:

# the same averages with RDDs: carry (sum, count) pairs yourself
(sc.textFile("people.tsv")
   .map(lambda line: line.split("\t"))
   .map(lambda x: (x[0], [int(x[1]), 1]))
   .reduceByKey(lambda x, y: [x[0] + y[0], x[1] + y[1]])
   .map(lambda x: [x[0], x[1][0] / x[1][1]])
   .collect())

Why [int(x[1]), 1]? reduceByKey combines partial results in any order, on any machine, like the combiner of lesson 30. Averages cannot be combined; (sum, count) pairs can, and the last map divides:

The DataFrame just says avg, and Spark supplies this machinery. (The slide's sc.textFile(...).split("\t") would fail: an RDD has no split, so each line is split inside a map.)

One engine, two front ends

Spark SQL is "part of the core distribution since Spark 1.0" and "runs SQL / HiveQL queries, optionally alongside or replacing existing Hive deployments" (p. 1270). Page 1272 shows where it sits.

The slide presents Spark SQL as "Originally called Shark" with three limitations (p. 1271): it "can only be used to query external data in Hive catalog → limited data sources", "can only be invoked via SQL string from Spark → error prone", and its "Hive optimizer tailored for MapReduce → difficult to extend". Read backwards, they are Spark SQL's design: any source and native RDDs, a DataFrame API, a new optimizer.

What a program has that an SQL string has not

Page 1273 lists the "Advantages over Relational Query Languages": "Holistic optimization across functions composed in different languages", "Control structures (e.g. if, for)" and "Logical plan analyzed eagerly → identify code errors associated with data schema issues on the fly". Execution is lazy, analysis is eager: on a table sales(region, amount) the line sales.select("col") fails at once, before any data is read, with ... `col` cannot be resolved. Did you mean one of the following? [`amount`, `region`] (Spark 4.2, executed).

Spark SQL also queries your own objects (p. 1274): it can "infer column names and types directly from data objects (via reflection in Java and Scala and data sampling in Python, which is dynamically typed)", reads them "in-place", and lets you "combine RDDs with external structured data", with "hot columns cached in memory".

Inside a full programming language a user-defined function (UDF) needs no packaging: Spark SQL "allows inline registration of UDFs", whereas Pig "requires the UDF to be written in a Java package", and a registered UDF is "available to other interfaces" (p. 1275) — the slide registers a trained model as predict and calls SELECT predict(age, weight) FROM users. Page 1269 adds composition: add_demographics takes a DataFrame and returns one. Runnable, with the caller of p. 1281:

from pyspark.sql import functions as F
from pyspark.sql.types import StringType

zipToCity = F.udf(lambda z: "Melbourne" if z.startswith("3")
                            else "Other", StringType())

def add_demographics(events):
    u = spark.table("users")
    return (events.join(u, events.user_id == u.user_id)
                  .withColumn("city", zipToCity(u.zip)))

# the timestamps of the events of users living in Melbourne
events = add_demographics(spark.read.parquet("/data/events"))
training_data = (events.where(events.city == "Melbourne")
                       .select(events.timestamp)
                       .collect())

Spark SQL is one structured API and one optimizer behind both SQL strings and DataFrame code, executed as RDD operations.

Let the optimizer do the hard work: Catalyst

A lambda is a black box: Spark can call it but cannot look inside. A DataFrame query is kept as a tree of node objects. The expression x + (1 + 2) is Add(Attribute(x), Add(Literal(1), Literal(2))), and a rule is a function that finds a pattern in the tree and replaces it (p. 1276):

tree.transform {
  case Add(Literal(c1), Literal(c2)) => Literal(c1+c2)
}

Rules (p. 1277) are "pattern matching functions that transform subtrees into specific structures", and partial functions: they "skip over subtrees that do not match → no need to modify existing rules when adding new types of operators" — which is why Catalyst is extensible. One transform may hold several patterns; rules come in batches and "may take multiple batches to reach a fixed point", where no rule changes the tree any more (one rewrite can enable the next); and "transform can contain arbitrary Scala code".

Four phases, from query to RDDs

Page 1278 is the map of the optimizer, and its caption is the fact to remember: "DataFrames and SQL share the same optimization/execution pipeline".

Analysis (p. 1279) turns the unresolved logical plan into a logical plan. For SELECT col FROM sales it must "look up relations by name from the catalog", "map named attributes to the input provided given operator's children", give a "UID for references to the same value" and "propagate and coerce types through expressions (e.g. 1 + col)". Spark lets you watch: for SELECT amount + 1 FROM sales the parsed plan holds 'amount and 'UnresolvedRelation [sales], the analysed plan amount#32 + cast(1 as double) — a column with a unique id, and a literal cast to its type.

Logical optimization (p. 1280) "applies standard rule-based optimization (constant folding, predicate-pushdown, projection pruning, null propagation, boolean expression simplification, etc)", all in "800LOC". Predicate pushdown moves a filter as close to the data as possible; projection pruning keeps only the columns used further up. Page 1282 spells out one such rule: "Find filters on top of projections", "Check that the filter can be evaluated without the result of the project", "If so, switch the operators".

Spark 4.2 goes further: its optimised plan drops the inner projection and turns the filter into isnotnull(id) AND (id = 1) (executed).

Page 1281 runs the slide's example through three plans (the widget replays them): the caller's where sits on top of the join written inside add_demographics, and Catalyst moves it down anyway — the "holistic optimization across functions" of p. 1273.

Physical planning turns the optimised plan into one or more physical plans of real operators, and a cost model selects one (p. 1278). In the Spark SQL paper it chose only the join algorithm — a broadcast join for a small table, as in the widget (background). The scans belong to data sources: "Spark's Data Source API allows optimizations like column pruning and filter pushdown into custom data sources" (p. 1268) — p. 1280's projection pruning and predicate pushdown. A source implements buildScan(requiredColumns: Array[String], filters: Array[Filter]): RDD[Row] (p. 1267): its two parameters are the two pushdowns. And since "DataFrame execution is done with Spark RDDs", custom code can drop to them: queryResult.rdd.mapPartitions { … }.

Code generation (p. 1283) "relies on Scala's quasiquotes": Catalyst "transforms a SQL tree into an abstract syntax tree (AST) for Scala code". The slide's generator is one pattern match:

def compile(node: Node): AST = node match {
  case Literal(value) => q"$value"
  case Attribute(name) => q"row.get($name)"
  case Add(left, right) =>
    q"${compile(left)} + ${compile(right)}"
}

Each q"…" is a code template. The tree of Fig. 2 becomes the line row.get("x") + 3, compiled once instead of walked for every row: for x+x+x a billion times, about 36 s interpreted against about 9 s generated, close to hand-written code (read off the chart), from a generator of "700LOC".

You say what; Catalyst decides how — it rewrites a tree with rules, chooses operators by cost and compiles the result.

Files that know their own shape: Avro and Parquet

Catalyst hands a scan requiredColumns and filters; what the scan does with them depends on the file. Page 1242 lists the "Available File Formats": text and CSV, JSON, SequenceFile ("binary key/value pair format"), Avro, Parquet, ORC ("optimized row columnar format") and "Data Frames". Three families hide in it: text (readable, slow to parse, no schema), binary rows (SequenceFile, Avro) and binary columns (Parquet, ORC).

Avro: rows, with the schema inside

Avro (p. 1243) is a "language neutral data serialization system": "write a file in python and read it in C". The schema is "language independent", "usually written in JSON", the data "encoded in binary format". It "supports schema evolution" — "producers and consumers at different versions of schema" — and "supports compression and are splittable". The tweets schema of p. 1244, repaired:

{
  "type": "record",
  "name": "tweets",
  "doc": "schema for storing tweets",
  "fields": [
    { "name": "username",  "type": "string" },
    { "name": "tweet",     "type": "string" },
    { "name": "timestamp", "type": "long" }
  ]
}

An Avro file (p. 1244) is a header — metadata holding the JSON schema, and a sync marker — then blocks: auxiliary data (record count and size), the serialized objects, and the same marker. That marker makes the file splittable: a task that starts mid-file, at an HDFS split, skips to the next marker and is aligned on a block. The schema is written once, so records carry no field names.

Parquet: columns inside row groups

"Apache Parquet is a columnar storage format available to any project in the Hadoop/Spark ecosystem, regardless of the choice of data processing framework, data model or programming language" (p. 1250). Page 1254 draws the difference on a table X, Y, Z: a row format stores x1 y1 z1 x2 y2 z2 …, a columnar format stores x1 … x5, then y1 … y5, then z1 … z5, each column as one "encoded chunk" — the column store of lesson 20, as a file.

The units nest (pp. 1245, 1251). A row group is "a group of rows in columnar format": the "max size buffered in memory while writing", "one (or more) per split while reading", "roughly: 50MB < row group < 1 GB". A column chunk is "the data for one column in a row group" and "can be read independently". A page is the "unit of access in a column chunk", "big enough for compression to be efficient", "roughly: 8KB < page < 1MB". The footer (p. 1252) "contains column chunks offset and schema": per column chunk its type, encodings, codec, number of values, offsets and sizes. It comes last, so the writer can stream row groups out and append the index at the end; a reader reads it first and jumps to the chunks it needs.

Mind the word block: p. 1245 calls a row group a block; p. 1260 means the HDFS block of lesson 31. No row group straddles two HDFS blocks — the slide packs two into one and pads the rest — so a task reads its row group from one machine.

Avro writes whole records one after another with the schema in its header; Parquet writes each column of a row group together and indexes the file in its footer.

Read less data: inside a Parquet file

Page 1257 lists three ways to "Read Less Data". Columnar organization: "Encoding: make the data smaller" and "Column projection: read only the columns you need". Row group filtering: "Use footer stats to eliminate row groups" and "Use dictionary pages to eliminate row groups". Page filtering: "Use page stats to eliminate pages". Page 1255 calls the first moves projection push down (read only the requested columns) and predicate push down (use statistics, such as each chunk's minimum and maximum, to skip what cannot match): together you "read only the data you need". The layout also "minimizes CPU cache misses": a column's values lie side by side, so the processor fetches only useful ones.

The dictionary of a column chunk (p. 1261) "is a compact list of all the values": "Search term missing? Skip the row group" — "like a bloom filter without false positives". Minimum and maximum rule a value out only when it lies outside their range; inside it they say "maybe". A dictionary answers exactly.

Encodings: make the data smaller

Before any compressor runs, Parquet encodes each column with a trick that fits its content (pp. 1256, 1258; the examples are ours).

Dictionary encoding is "useful for columns with few ( < 50,000 ) distinct values" and "when applicable, compresses better and faster than heavyweight algorithms". Bit packing stores small integers "in the minimum bits required": a value whose maximum is \(m\) needs \(\lceil \log_2(m+1) \rceil\) bits, so 16 bits cover any key below 65,536. Parquet 2.0 (p. 1259) adds "more encodings: compact storage without heavyweight compression" (delta encodings "for integers, strings and sorted dictionaries"), "statistics: to be used by query planners and predicate pushdown" and a "new page format: to facilitate skipping ahead at a more granular level".

Nested records in flat columns

Parquet's "key strength is to store nested data in truly columnar format using definition and repetition levels" (p. 1254). The technique comes from Google's Dremel; p. 1253 applies it to a schema with an optional group and two repeated fields.

Next to each stored value Parquet writes two small integers. The definition level D says how many of the optional or repeated fields on the value's path are actually present; the repetition level R says at which repeated field the value continues a list, 0 meaning that a new record starts. The slide's record, shredded step by step:

1. Read the columns off the schema, one per leaf: DocId, Links.Backward, Links.Forward. The groups Document and Links store nothing themselves.

2. Count, on each path, the fields that may be missing:

   \[ D_{\max} = \#\ \text{optional or repeated fields} \]

   DocId is required: 0. Links is optional and Backward repeated: 2; Forward too. A value must say how deep its path is defined.

3. Count only the repeated fields:

   \[ R_{\max} = \#\ \text{repeated fields} \]

   0 for DocId, 1 for each list: with one repeated field on the path, R is 0 or 1.

4. Shred DocId: 20: a required value at the start of a record gives (20, R 0, D 0).

5. Shred Backward: 10, 30. The first value starts the record's list (R 0) with Links and Backward present (D 2): (10, 0, 2). The second continues the list at Backward, the first repeated field: (30, 1, 2).

6. Shred Forward: 80, the record's first Forward: (80, 0, 2). These are the four rows of p. 1253; with maxima 1 and 2, bit packing stores R in 1 bit and D in 2.

7. Not on the slide, same rules: a record without Links stores NULL in both Links columns with R 0, D 0 — the path stops before Links. R = 0 tells the assembler a new record begins, so it rebuilds the record exactly.

Parquet skips at three levels — column chunks, row groups, pages — and turns nesting into two tiny integers per value.

Beyond queries: extensions, JSON, pipelines, speed

Two "Extensions" complete the design (p. 1284). A new data source "must implement a createRelation function that takes a set of key-value params and returns a BaseRelation object" — "E.g. CSV, Avro, Parquet, JDBC". A user-defined type (UDT) maps "user-defined types to structures composed of Catalyst's built-in types": PointUDT stores a Point as a struct of two doubles, x and y, with serialize and deserialize. Catalyst never needs to understand a Point, only the struct.

Schema inference for JSON

JSON carries no schema, yet Spark SQL "automatically infers schema from a set of records, in one pass or sample" (p. 1285). Each record yields "a tree of STRUCT types"; each field gets "the most appropriate type … based on all data observed in that column"; the per-record schemas are merged "in one reduce operation" — possible because merging schemas is associative, like the (sum, count) pairs. The slide's three tweets:

An integer in one record and a fraction in another make FLOAT, the most specific type holding both; loc is missing from record 3, so it gets no NOT NULL. Spark 4.2 reads the same records as struct<lat: double, long: double>, every field nullable (executed): the slide's reasoning, newer type names. "Same trick for Python typing" is the data sampling of p. 1274, and all of it is the schema-on-read of lesson 4.

ML pipelines speak DataFrame

Page 1286, repeated on p. 1288, chains three stages of Spark's machine-learning library, MLlib (quotes straightened):

from pyspark.ml import Pipeline
from pyspark.ml.feature import Tokenizer, HashingTF
from pyspark.ml.classification import LogisticRegression

tokenizer = Tokenizer(inputCol="text", outputCol="words")
hashingTF = HashingTF(inputCol="words", outputCol="features")
lr = LogisticRegression(maxIter=10, regParam=0.01)
pipeline = Pipeline(stages=[tokenizer, hashingTF, lr])
# df: a "text" and a "label" column; returns a PipelineModel
model = pipeline.fit(df)

tokenizer and hashingTF only transform; lr is an estimator, whose fit on ds2 produces the transformer lr.model. That is why lr sits outside the dotted "Pipeline Model" box and lr.model inside (executed). If you have used scikit-learn pipelines, fit and transform mean the same here.

The twelve times, explained

Page 1287 measures the promises (Fig. 8).

Back to the opening puzzle. Lambdas are black boxes that Spark can only call, row by row, in a Python process; the one-liner is a tree it can read, and that gives three answers. Catalyst can rewrite it. Only the plan is built in Python and the work runs as JVM bytecode: the paper credits this for the 12×, and code generation, which "avoids expensive allocation of key-value pairs", for the 2× over Scala. And the plan names its columns and filters, so the data source can be asked for less: with Parquet, only the needed column chunks and row groups leave the disk — the answer in the file. The right chart adds that one DataFrame job beats an SQL query plus a separate Spark job by about 2×: nothing is saved to HDFS in between (paper).

The DataFrame wins because Spark sees the whole plan: it can rewrite it, compile it and ask the file for less.