DataFrames & Datasets: The Structured APIs

A DataFrame is a table that Spark can spread over many machines. You met it in the chapter "What Apache Spark Is & Why It Exists". In this chapter you look closer at how it works.

From "Transformations, Actions & Lazy Evaluation" you know transformations, actions, and explain(). Here you learn the parts you use in every Spark job.

A DataFrame also has partitions, the pieces you saw in the architecture chapter. You think in rows and columns. Spark runs the work on partitions.

• The schema: the names and types of the columns.
• Columns and expressions: select, withColumn, when, and cast.
• The same query in the DataFrame API and in SQL.
• NULL values: what sum, avg, and count do with them.
• Datasets, Row objects, and toPandas().

The data is the shop. The table orders has six rows, orders 101 to 106. Here they are.

orders.show()

The schema is the list of columns in a DataFrame. For each column it gives the name and the type. A type says what kind of value the column holds, such as a whole number or text.

Spark needs the schema to plan your work. Call printSchema() to see it.

orders = spark.read.csv("orders.csv", header=True, inferSchema=True)
orders.printSchema()

Guessing is risky in a pipeline. A pipeline is a job that runs again and again on new files. The guess can change when the data changes.

You can give the schema instead. The short way is a DDL string.

DDL means data definition language. A DDL string is a list of "name type" pairs.

orders = spark.read.csv(
    "orders.csv",
    header=True,
    # Python joins the two strings into one
    schema="order_id INT, customer_id INT, status STRING, "
           "amount DECIMAL(10,2), order_date DATE",
)
orders.printSchema()
orders.show()

The long way gives the same result. It uses StructType, a list of StructField objects. Each StructField has a name, a type, and a nullable flag.

from pyspark.sql.types import (
    StructType, StructField, IntegerType, StringType, DecimalType, DateType,
)

schema = StructType([
    StructField("order_id", IntegerType(), True),
    StructField("customer_id", IntegerType(), True),
    StructField("status", StringType(), True),
    StructField("amount", DecimalType(10, 2), True),
    StructField("order_date", DateType(), True),
])
orders = spark.read.csv("orders.csv", header=True, schema=schema)

An explicit schema is safer for three reasons.

• Spark does not need an extra pass over the file to guess the types.
• The types do not change when the data changes.
• You choose exact types, such as decimal for money.

F is pyspark.sql.functions, imported as F, as in the chapter "What Apache Spark Is & Why It Exists". F.col("amount") points at a column. An expression is a calculation built from columns, such as F.col("amount") - 5.

Spark does not run it yet. It is a recipe.

select() keeps the columns you name. It can also make new ones. Here select() keeps order_id and calculates a net amount.

from pyspark.sql import functions as F

orders.select(
    "order_id",
    (F.col("amount") - 5).alias("net"),
).show()

withColumn() adds one column and keeps all the others. It does not change the old DataFrame. It returns a new one.

F.when() is a CASE WHEN: it picks a value by a test. cast() changes the type.

labelled = (
    orders
    .withColumn("amount_int", F.col("amount").cast("int"))
    .withColumn(
        "size",
        F.when(F.col("amount") >= 50, "big").otherwise("small"),
    )
    .select("order_id", "amount", "amount_int", "size")
)
labelled.show()

You can also write SQL. First, give the DataFrame a name that SQL can use.

That name is a temp view. It lives only in your session and copies no data.

Then spark.sql() runs a SQL query on it and returns a DataFrame. Both versions below ask the same question: how much did each customer pay on paid orders?

# DataFrame API
revenue_api = (
    orders
    .filter(F.col("status") == "paid")
    .groupBy("customer_id")
    .agg(F.sum("amount").alias("revenue"))
    .orderBy("customer_id")
)

# SQL on a temp view
orders.createOrReplaceTempView("orders")
revenue_sql = spark.sql("""
    SELECT customer_id, SUM(amount) AS revenue
    FROM orders
    WHERE status = 'paid'
    GROUP BY customer_id
    ORDER BY customer_id
""")
revenue_sql.show()

Catalyst is the part of Spark that makes your plan cheaper before it runs. You met its work in the lazy evaluation chapter.

Both versions go to Catalyst. In simple queries like this one, they usually get the same plan. Check it: call revenue_api.explain() and revenue_sql.explain(), and compare.

Both are fine. Pick the style your team reads best.

NULL means "no value". Spark follows the same NULL rules as SQL.

A calculation with NULL gives NULL. A test with NULL is not true.

The aggregate functions skip NULL. That changes the result. Look at order 105 and the amount column.

orders.agg(
    F.count("*").alias("all_rows"),
    F.count("amount").alias("amounts"),
    F.sum("amount").alias("total"),
    F.round(F.avg("amount"), 2).alias("average"),
).show()

The average divides by 5, not 6. Order 105 is not counted at all.

Often this is what you want. Check that it is true for your report.

To treat NULL as a value, replace it. F.coalesce() returns the first value that is not NULL.

The method na.fill() is a short way to do the same for the columns you name.

filled = orders.withColumn(
    "amount_filled", F.coalesce(F.col("amount"), F.lit(0))
)
filled.select("order_id", "amount_filled").show()

# The same fix for one column:
filled_too = orders.na.fill(0, subset=["amount"])

filled.agg(F.round(F.avg("amount_filled"), 2).alias("average")).show()

In the architecture chapter, collect() brought every row to the driver. Now look at what you get. It is a Python list of Row objects.

A Row is one row of the DataFrame, as a Python object. You read a value by name, with row["amount"] or row.amount.

paid = (
    orders
    .filter(F.col("status") == "paid")
    .select("order_id", "amount")
    .orderBy("order_id")
    .collect()
)
print(paid[0])
print(paid[0]["amount"])
print(paid[0].order_id)
print(len(paid))

After collect(), the data is plain Python on the driver. Spark does not help you any more. Use collect() only on small results, such as a total.

A Dataset is a DataFrame with a Java or Scala type for each row, such as an Order class. The compiler then checks your column names before the job runs.

Datasets exist only in Scala and Java. PySpark has DataFrames only. In Python you do not use Datasets, and that is fine.

An RDD (resilient distributed dataset) is Spark's older, lower-level API. It is a list of objects split over partitions, with no columns. The compiler is the program that checks Scala or Java code before it runs.

In Scala, a DataFrame is a Dataset of Row objects. In PySpark, a wrong column name shows up as an error when Spark builds the plan. Python cannot catch it earlier.

pandas is a Python library for tables on one machine. toPandas() turns a Spark DataFrame into a pandas DataFrame.

It works like collect(). Every row moves to the driver. Use it only after you make the data small.

small = (
    orders
    .filter(F.col("status") == "paid")
    .select("order_id", "amount")
    .toPandas()
)
print(small)

You can add .limit(1000) before toPandas() as a safety limit.

• The schema lists column names and types. Read it with printSchema().
• In pipelines, write the schema yourself. A DDL string is the short way.
• select, withColumn, F.when, and cast build new columns from expressions.
• The DataFrame API and spark.sql() both go to Catalyst. Pick the style your team reads best.
• sum, avg, and count("amount") skip NULL. count("*") does not. Fill NULL with 0 only when 0 is the right value.
• Datasets are Scala and Java only. collect() gives a list of Row. toPandas() moves all rows to the driver.

Try these without looking back.

1. Which average do you get for orders: 44.10 or 36.75? Which line of code gives the other one?
2. Write the DDL string for a table with order_id as a whole number and amount with 2 decimals.
3. Why does order 105 get "small" from F.when(F.col("amount") >= 50, "big").otherwise("small")?