Spark DataFrames and Spark SQL

Data Engineering & Modern Data Platforms · Lesson 27 · Big Data Processing and Apache Spark

Lesson Overview

Apache Spark Architecture named the driver, the executors, and the moment an action becomes a job. This lesson is how you describe the progress landing so that plan is worth running. A DataFrame is a distributed collection of rows with a schema. Spark SQL is the same engine with a query string. Both build a plan the Catalyst optimizer can rewrite: drop unused columns, push a predicate toward a Parquet file, and turn a filter-plus-project into one stage. A Python loop over the rows cannot be rewritten, because the engine cannot see inside it.

TechLearningPro's landing still accepts the contract from Introduction to Data Engineering: event_id, learner_id, lesson_id, event_type, attempt_number, event_time, and an optional checkpoint. Identical payloads collapse. Conflicting payloads refuse the run. Prohibited fields refuse the file. The publish is events, partitioned by UTC event_date, and completion-v3 stays SQL in the warehouse sense over that publish. This lesson writes the landing's transformations and the actions that force them. It does not yet package a scheduler, a checkpoint file, and a committer configuration. That packaging is PySpark Data Engineering. The expressions here are the ones that job will call.

PySpark is the surface. The work runs in the executor JVM when you stay on column functions. A Python UDF moves each row to a worker process, which the architecture lesson already priced. The examples pin the APIs that have been stable through PySpark 3.5: SparkSession, StructType, pyspark.sql.functions, and explain. Where a default might differ on a later build, the session sets it explicitly. /learn/python is the driver language. /learn/sql is the dialect Spark SQL tracks closely enough that a SELECT in this lesson should be readable, and different enough that types and nulls still need a test.

Learning Objectives

By the end of this lesson you will be able to:

1. Define a DataFrame as rows plus a schema, and explain why nullability on that schema is not a database constraint.
2. Separate transformations from actions, and name which landing steps must be actions.
3. Declare the progress schema and refuse inferred types on a production read.
4. Detect unknown and prohibited JSON keys, corrupt lines, naive timestamps, and conflicting event_id payloads with column expressions.
5. Read an explain plan for a scan, a project, and an exchange, and relate the exchange to a stage.
6. Use a temporary view and Spark SQL for a check a reviewer wants to read as a query, without a second engine.
7. Keep dropDuplicates and "latest row" windows from becoming the collision policy.

Prerequisites

• Apache Spark Architecture: jobs, stages, tasks, driver versus executor, and the cost of a Python UDF. This lesson assumes that map and only adds the DataFrame vocabulary.
• Big Data Fundamentals: the publish schema and event_date as the directory partition. The decision to use Spark at all should already be written.
• Data Formats and Serialization: JSONL input, Parquet output, and why a declared schema beats a sample.
• Building Batch Data Pipelines with Python: the canonical-hash collision rule this lesson reimplements with a shuffle instead of a dict.
• Lessons 6 and 7 and /learn/sql: GROUP BY, COUNT, and nulls. Spark SQL is not PostgreSQL. Test casts.
• PySpark 3.5 and Java 17 if you execute the examples. local[2] is enough. No cluster.

Detailed Core Concepts

Rows, columns, and a schema you can defend

A DataFrame is immutable. withColumn returns a new DataFrame. The variable you still hold refers to the old plan. That is why landing code reads better as a function that takes a DataFrame and returns a DataFrame than as a script that mutates a table in place. There is no in-place update of a progress row inside Spark. A correction is a new publish of the date partition, which matches the replace rule you already have.

The schema is a StructType of fields with names and data types. Production reads of progress pass that schema to the reader. Inference samples rows and guesses. A sample where attempt_number is always a small integer may become IntegerType on Tuesday and LongType on a day a broken exporter sends a larger number, or a string on a day it arrives quoted. Downstream completion-v3 then sees a type change in a column that was supposed to be stable. Inference is acceptable in a notebook that is about to print ten rows and be thrown away. It is not acceptable on the landing.

Nullability flags on fields document intent. Spark does not enforce nullable=False as a constraint during a JSON read. A JSON null can become a SQL null in a non-nullable field and travel until a later expression or an ANSI-mode cast objects. The landing therefore checks required columns explicitly and fails the line or the file according to the contract. The schema is the type contract. The checks are the business contract. You want both.

spark.sql.ansi.enabled set to true makes many illegal casts throw instead of quietly becoming null. The session in this lesson turns it on so a boolean slipped into attempt_number does not become a polite null that a loose filter might drop or, worse, a coerced one. You still test the fixture. ANSI mode is a session setting, not a substitute for the prohibited-field rule.

Transformations and actions

Transformations add steps to the plan. On a DataFrame they include select, filter (and where), withColumn, drop, groupBy, join, dropDuplicates, repartition, and unionByName. Actions force a job. They include count, collect, take, show, head, toLocalIterator in the sense that it pulls rows, and any write. explain asks the driver to print a plan and does not scan the dataset.

The landing wants as few actions as the contract allows. A useful shape is: build the plan that parses and classifies lines; action one materializes a small summary (accepted count, rejected count, whether any prohibited key exists, whether any event_id has two hashes); if the summary says refuse, stop without a publish write; if it says publish, action two writes Parquet. Two actions mean two jobs, and the parse may run twice unless you persist the intermediate DataFrame. Persisting a narrow projection (id, hash, flags) can be worth it. Persisting every raw line usually is not. Measure it on a slice. The architecture lesson's cache warning still applies.

show in a production driver is an action that pulls rows to the driver and prints them into logs. That is a payload leak and a hidden job. Tests may collect a fixture of five rows. The landing logs counts.

Narrow and wide, in DataFrame words

select, withColumn, and filter on a row's own columns are narrow. They pipeline into the scan stage. groupBy("event_id") is wide. The plan contains an exchange. repartition("event_id") is a shuffle you asked for by name. dropDuplicates(["event_id"]) shuffles, and it keeps an arbitrary survivor. Arbitrary is not the contract. The contract is: one canonical hash, collapse duplicates; two hashes, fail. dropDuplicates cannot say "fail." It can only say "I kept one." Use it after the conflict check has passed, to collapse identical hashes, or use a countDistinct and never let a conflict reach the write.

A window that does row_number ordered by event_time and keeps row 1 is the "latest wins" bug with extra syntax. Do not import it from a deduplication blog. Two payloads are a refusal, even if one of them is later.

Catalyst, in the amount an operator needs

When you call a transformation, the driver holds an unresolved logical plan: column names that might be wrong, types that might not line up. Analysis binds names to the schema and rejects a column you do not have. Optimization applies rules: predicate pushdown, projection pruning, constant folding, filter combination. The physical plan chooses operators, including hash aggregates and sort aggregates, and exchanges. explain(mode="formatted") shows that physical plan. explain(mode="extended") shows the earlier trees when you need to know whether a filter survived.

For the landing, the practical checks are: the JSON or Parquet scan lists only the columns you need; a filter on a partition column is a partition prune when the data is already partitioned; the collision groupBy is an exchange of the hash and the id, not of the whole line; no Python UDF node sits on the per-row path. If the formatted plan shows a BatchEvalPython or similar Python evaluation node on the hot path, the landing is paying the worker boundary. Replace that expression with a built-in function.

Spark SQL and the DataFrame API share this optimizer. createOrReplaceTempView registers a name. spark.sql parses a query against it. The view sees the same plan. Use SQL when a reviewer of the collision rule wants to read GROUP BY. Use the DataFrame API when the landing is a function composed in Python modules. Do not maintain two copies of the rule, one in each style. One implementation, and a SQL string is fine if that is the implementation.

JSON is text until you impose a schema

spark.read.json with a schema parses known fields and, by default, drops unknown fields. That drop is the opposite of lesson 14, which failed the file on unknown keys so email could not arrive under a new name. A production landing therefore reads lines with spark.read.text, looks at the keys, and only then parses the object.

json_object_keys, a SQL function since Spark 3.1, returns the top-level keys of a JSON object. Progress events are flat, so top-level keys are the contract surface. A key outside the allowed set is a refusal reason: email, name, and payment_token are prohibited specifically, and any other stranger is unknown. The file-level decision is an action that asks whether any such row exists. Quarantine is for malformed JSON, missing required fields, bad types, and timestamps without an offset. Those rows are written aside, under the cap, and are not published.

from_json with a schema parses a column. Mode PERMISSIVE puts a corrupt record into a column you must declare, named by columnNameOfCorruptRecord. Mode FAILFAST fails the task, which fails the run, which is correct for a breach and harsh for one bad line. Mode DROPMALFORMED drops the line with no remainder. DROPMALFORMED is the silent-success anti-pattern. This lesson uses permissive parsing plus an explicit remainder table, and a separate key check that permissive mode will not do for you.

Dates, partitions, and predicate pushdown

event_time stays a UTC instant rendered as text, as in the Python landing, so the published column does not depend on a session zone. event_date is yyyy-MM-dd derived from that instant and is the partition column. Deriving it in the landing, once, stops every reader from reinterpreting a naive timestamp. Lines without an offset are rejects. The session timezone is set to UTC anyway, so a bug that forgets the offset check does not silently use the driver's local zone. The check remains. Defense in one layer is how Tuesday's laptop and Friday's cluster disagree.

When the published Parquet is read later, a filter on event_date prunes directories. A filter on lesson_id does not prune directories, because lesson_id is a column. It can still skip Parquet row groups if statistics say the value is absent. That is the lesson 11 behavior Spark's reader implements. Selecting only learner_id and lesson_id prunes the other columns at the read. Both wins disappear if you select("*") and then convert to RDD of full rows "for flexibility."

What the driver is allowed to see

count, agg(...).collect() of a few metrics, and limit(1).count() to test emptiness of the conflict set are driver-sized results. The conflict check's shuffle still scans the data. limit does not make a group-by lazy in the sense of reading one row. It limits the output after the aggregate. Expect a full pass. take(20) on rejected reasons during a test is fine. take(20) on accepted events in production logs learner activity. Log the count of rejects and a stable reason code, not the line.

Architecture and Design Diagrams

The landing plan is a narrow classify stage and a wide identity stage. Writes are actions that sit after the decision, not inside the classification.

mermaidCopy
flowchart LR
    text[Text lines of JSONL]
    keys[Key check and from_json]
    narrow[Project, time normalize, classify]
    wide[Aggregate distinct hashes per event_id]
    decide[Driver reads a small summary]
    refuse[No write]
    publish[Write accepted Parquet]
    quarantine[Write reject reasons]

    text --> keys --> narrow --> wide --> decide
    decide -->|breach or over cap| refuse
    decide -->|within contract| publish
    decide -->|ordinary rejects| quarantine

Catalyst sits between the DataFrame calls and the tasks. You read the physical plan. You do not execute the Python as if each call were a stage.

mermaidCopy
flowchart TD
    api[DataFrame calls or Spark SQL]
    unresolved[Unresolved logical plan]
    analyzed[Analyzed plan bound to the schema]
    optimized[Optimized plan]
    physical[Physical plan with exchanges]
    jobs[Jobs only when an action runs]

    api --> unresolved --> analyzed --> optimized --> physical --> jobs

SQL and the DataFrame API are two doors to that same sequence.

mermaidCopy
flowchart LR
    df[DataFrame transforms in Python modules]
    view[Temporary view]
    sql[spark.sql review query]
    cat[Catalyst]
    df --> cat
    df --> view --> sql --> cat

Technical Explanation

A declared schema

The fields match lesson 24, with the corrupt-record column available when a reader uses permissive mode on a struct parse. The text-first path below parses with from_json into the contract struct and keeps classification flags beside it.

pythonCopy
from pyspark.sql.types import IntegerType, StringType, StructField, StructType

CONTRACT_FIELDS = (
    "event_id",
    "learner_id",
    "lesson_id",
    "event_type",
    "attempt_number",
    "event_time",
    "checkpoint",
)
PROHIBITED = ("email", "name", "payment_token")
ALLOWED = CONTRACT_FIELDS

PROGRESS_STRUCT = StructType(
    [
        StructField("event_id", StringType(), nullable=False),
        StructField("learner_id", StringType(), nullable=False),
        StructField("lesson_id", StringType(), nullable=False),
        StructField("event_type", StringType(), nullable=False),
        StructField("attempt_number", IntegerType(), nullable=False),
        StructField("event_time", StringType(), nullable=False),
        StructField("checkpoint", IntegerType(), nullable=True),
    ]
)

checkpoint the field is the player's integer, not the landing checkpoint. The name collision is inherited from the event contract. In code, call the business file landing_checkpoint so a review does not mix them.

Classifying a line

The next function is a transformation. It does not read data until an action. It adds columns a later filter can split on: reject_reason, file_breach, and a canonical hash for rows that parsed.

pythonCopy
from pyspark.sql import DataFrame
from pyspark.sql import functions as F

def classify_lines(lines: DataFrame) -> DataFrame:
    allowed = F.array(*[F.lit(name) for name in ALLOWED])
    prohibited = F.array(*[F.lit(name) for name in PROHIBITED])
    keys = F.json_object_keys(F.col("value"))
    extra = F.array_except(keys, allowed)
    banned = F.array_intersect(keys, prohibited)
    parsed = F.from_json(F.col("value"), PROGRESS_STRUCT, {"mode": "PERMISSIVE"})
    event_type_ok = parsed["event_type"].isin("checkpoint", "completed")
    has_offset = parsed["event_time"].rlike(r"(Z|[+-][0-9]{2}:[0-9]{2})
quot;)
    instant = F.to_timestamp(parsed["event_time"])
    event_time_utc = F.date_format(instant, "yyyy-MM-dd'T'HH:mm:ss'Z'")
    required_missing = (
        parsed["event_id"].isNull()
        | parsed["learner_id"].isNull()
        | parsed["lesson_id"].isNull()
        | parsed["event_type"].isNull()
        | parsed["attempt_number"].isNull()
        | parsed["event_time"].isNull()
    )
    breach = keys.isNotNull() & ((F.size(banned) > 0) | (F.size(extra) > 0))
    reason = (
        F.when(keys.isNull(), F.lit("malformed_json"))
        .when(F.size(banned) > 0, F.lit("prohibited_field"))
        .when(F.size(extra) > 0, F.lit("unknown_field"))
        .when(required_missing, F.lit("missing_or_bad_type"))
        .when(~event_type_ok, F.lit("bad_event_type"))
        .when(~has_offset | instant.isNull(), F.lit("bad_timestamp"))
        .when(parsed["attempt_number"] < 1, F.lit("bad_attempt"))
        .otherwise(F.lit(None))
    )
    canonical = F.sha2(
        F.to_json(
            F.struct(
                parsed["event_id"].alias("event_id"),
                parsed["learner_id"].alias("learner_id"),
                parsed["lesson_id"].alias("lesson_id"),
                parsed["event_type"].alias("event_type"),
                parsed["attempt_number"].alias("attempt_number"),
                event_time_utc.alias("event_time"),
                parsed["checkpoint"].alias("checkpoint"),
            )
        ),
        256,
    )
    return lines.select(
        parsed["event_id"].alias("event_id"),
        parsed["learner_id"].alias("learner_id"),
        parsed["lesson_id"].alias("lesson_id"),
        parsed["event_type"].alias("event_type"),
        parsed["attempt_number"].alias("attempt_number"),
        event_time_utc.alias("event_time"),
        parsed["checkpoint"].alias("checkpoint"),
        F.date_format(instant, "yyyy-MM-dd").alias("event_date"),
        canonical.alias("canonical"),
        reason.alias("reject_reason"),
        breach.alias("file_breach"),
    )

Several details are deliberate. to_json on a struct with named fields fixes key order for the hash, so two encodings of the same contract hash together. The hash covers the UTC instant, not the raw string, so Z and +00:00 collapse the way lesson 14 collapsed them. file_breach is true when prohibited or unknown keys exist, even if from_json would have dropped those keys and produced a clean struct. That is the whole point of json_object_keys. Rows with a null key array are malformed JSON, a line reject, not proof that the file is free of email. A file that fails to parse is quarantined per line. A file that parses and contains email is a breach.

rlike and to_timestamp are built-in. They stay in the JVM. A Python function that calls datetime.fromisoformat per row would also be correct and would show up as a Python worker in the plan. Prefer the built-in. If a fixture disagrees with to_timestamp on an unusual offset, fix the pattern in the built-in parser or reject the line. Do not paper over it with a UDF until you have a case the engine cannot represent.

The collision aggregate

pythonCopy
def conflict_keys(classified: DataFrame) -> DataFrame:
    good = classified.where(F.col("reject_reason").isNull() & ~F.col("file_breach"))
    return (
        good.groupBy("event_id")
        .agg(F.countDistinct("canonical").alias("versions"))
        .where(F.col("versions") > 1)
        .select("event_id")
    )

This is the wide step. The shuffle key is event_id. The payload is the hash, because countDistinct on canonical does not need learner_id. If you groupBy the whole struct, you shuffle the world. Project first. The filter that removes rejects before the aggregate is narrow and should sit under the exchange in the plan. Confirm that with explain.

Identical retries share a canonical hash, so versions stays 1 and the row is absent from conflict_keys. The publish path then keeps one copy:

pythonCopy
def accepted_events(classified: DataFrame) -> DataFrame:
    good = classified.where(F.col("reject_reason").isNull() & ~F.col("file_breach"))
    return good.dropDuplicates(["event_id", "canonical"]).select(
        "event_id",
        "learner_id",
        "lesson_id",
        "event_type",
        "attempt_number",
        "event_time",
        "checkpoint",
        "event_date",
    )

dropDuplicates is only valid once conflict_keys is empty. If you skip the conflict action and only drop duplicates, a two-payload id publishes one of them. Call the conflict action first. In lesson 28 the driver orders those actions and skips the write on failure.

A SQL form of the same check

Register the classified DataFrame and let a reviewer read the rule. This query is the check. It is not a second definition maintained by hand if you generate it from the same DataFrame.

pythonCopy
def conflicts_as_sql(spark, classified: DataFrame) -> DataFrame:
    classified.createOrReplaceTempView("classified_progress")
    return spark.sql(
        """
        SELECT event_id
        FROM classified_progress
        WHERE reject_reason IS NULL
          AND NOT file_breach
        GROUP BY event_id
        HAVING COUNT(DISTINCT canonical) > 1
        """
    )

The view is session-scoped. It disappears when the session does. It is not a warehouse table and not something completion-v3 should point at. Name it so nobody persists it with a saveAsTable out of curiosity. saveAsTable writes a catalog entry. This lesson does not use it for the landing.

Partition column and empty dates

event_date is null when the timestamp did not parse. Those rows already have reject_reason set, so they are absent from accepted_events. Still filter event_date non-null on the write DataFrame in the job, in case a later refactor forgets. A null partition column becomes a directory you did not intend, or a commit that does not line up with the date you meant to replace.

Actions that implement refuse-or-publish

The decision itself is an action, because the driver must branch. Branching in the driver on a collected flag is normal. Branching per row belongs in when. Do not collect the conflict ids into the driver beyond a single example id for the log, and even that id is an internal identifier. One row is enough to prove the set is non-empty:

pythonCopy
def has_file_breach(classified: DataFrame) -> bool:
    hit = classified.where(F.col("file_breach")).select(F.lit(1)).limit(1)
    return hit.count() == 1

limit(1) plus count still launches a job, and on a text scan it can stop early only if the plan allows a limit pushdown. A breach on the last line may scan the file. That is acceptable. A breach must not publish. The cost is one read, which you were going to pay anyway.

Reject counts:

pythonCopy
def reject_count(classified: DataFrame) -> int:
    row = (
        classified.where(F.col("reject_reason").isNotNull() & ~F.col("file_breach"))
        .agg(F.count("*").alias("n"))
        .collect()[0]
    )
    return int(row["n"])

Compare n to the cap from configuration. The cap is not a constant buried in classify_lines, so Tuesday's policy and Wednesday's policy do not require an edit to the expression. Prohibited fields ignore the cap. They are file_breach, not rejects.

Practical Implementation or Design Walkthrough

Problem

Implement the classify, conflict, and accept functions against a fixture of JSONL lines in local[2]. Assert the plan contains an exchange on the conflict DataFrame and that a prohibited field is a breach even though it is not in the struct. This is the transformation the production job will call. It does not write to HDFS.

Session and fixture

pythonCopy
import json
from pathlib import Path

from pyspark.sql import SparkSession

def lesson_session() -> SparkSession:
    return (
        SparkSession.builder.master("local[2]")
        .appName("tlp-dataframe-lesson")
        .config("spark.ui.enabled", "false")
        .config("spark.sql.shuffle.partitions", "4")
        .config("spark.sql.adaptive.enabled", "false")
        .config("spark.sql.session.timeZone", "UTC")
        .config("spark.sql.ansi.enabled", "true")
        .getOrCreate()
    )


def write_fixture(directory: Path) -> None:
    lines = [
        {
            "event_id": "e1",
            "learner_id": "l1",
            "lesson_id": "les1",
            "event_type": "checkpoint",
            "attempt_number": 1,
            "event_time": "2026-03-02T14:02:11Z",
            "checkpoint": 3,
        },
        {
            "event_id": "e1",
            "learner_id": "l1",
            "lesson_id": "les1",
            "event_type": "checkpoint",
            "attempt_number": 1,
            "event_time": "2026-03-02T14:02:11+00:00",
            "checkpoint": 3,
        },
        {
            "event_id": "e2",
            "learner_id": "l1",
            "lesson_id": "les1",
            "event_type": "completed",
            "attempt_number": 1,
            "event_time": "2026-03-02T15:00:00Z",
        },
        {"event_id": "e3", "event_type": "checkpoint"},
        {"event_id": "e4", "email": "hidden@example.com", "event_type": "checkpoint"},
    ]
    path = directory / "progress.jsonl"
    with path.open("w", encoding="utf-8") as handle:
        for line in lines:
            handle.write(json.dumps(line) + "\n")
        handle.write("{not json\n")

The duplicate e1 lines differ in the offset spelling and must become one canonical hash. e3 is a line reject. The email line is a file breach even if other fields are missing. The last line is malformed.

Assertions that encode the contract

pythonCopy
def run_fixture(spark: SparkSession, path: str) -> None:
    lines = spark.read.text(path)
    classified = classify_lines(lines).withColumn(
        "file_breach", F.coalesce(F.col("file_breach"), F.lit(False))
    )
    breaches = classified.where(F.col("file_breach"))
    if breaches.limit(1).count() != 1:
        raise AssertionError("email must be a file breach")
    reasons = {
        row["reject_reason"]
        for row in classified.select("reject_reason").distinct().collect()
        if row["reject_reason"] is not None
    }
    if "malformed_json" not in reasons:
        raise AssertionError("corrupt line must be visible")
    if "prohibited_field" not in reasons:
        raise AssertionError("prohibited field missing from classification")
    # A second fixture, with no email, is what production would be allowed to publish.
    # This file is a breach: stop before any accept write.
    conflict = spark.createDataFrame(
        [("e9", "a" * 64), ("e9", "b" * 64)],
        ["event_id", "canonical"],
    )
    versions = conflict.groupBy("event_id").agg(
        F.countDistinct("canonical").alias("versions")
    )
    versions.explain(mode="formatted")
    if versions.where(F.col("versions") > 1).count() != 1:
        raise AssertionError("two hashes must conflict")

The if False branch is a reminder I will not leave in the production module: do not "clean" a breach by filtering the string email and publishing the rest. The fixture function above is clearer if the breach assertion returns before any accept write, and a second fixture file without email feeds conflict_keys and accepted_events. Split the fixtures that way in the repository. One file, one expected decision.

On the clean duplicate of e1, accepted_events after an empty conflict_keys yields one row, event_time ending in Z, and event_date equal to 2026-03-02. Assert that with collect on the fixture only.

Call conflict_keys(...).explain(mode="formatted") and read the exchange. If someone rewrites the check as a Python UDF over groupBy, the plan changes and the test that the physical plan contains an Exchange and no Python eval on the classify path is the regression test. String-matching an explain plan is brittle across Spark versions. Prefer a small helper that checks str(plan).lower() for "exchange" in a test you are willing to update when you upgrade Spark, and keep the row-level assertions as the stable contract. The row assertions are the specification. The plan assertion is a performance guard.

What this walkthrough does not do

It does not write.parquet. It does not set a committer. It does not compute completion. It does not read hdfs://. Those are real and they are next. A DataFrame that is correct on five lines is the right first proof. A DataFrame that is correct on five lines and then collects the production path is not a proof of scale. Keep the actions in tests on fixtures, and keep the production actions on counts and writes.

Reading the published side, later

Once a date is published, a reader uses the same struct plus event_date and the basePath pattern from lesson 25. A filter event_date = '2026-03-02' should prune. explain on that read should show the partition filter. If it shows a scan of every date, the path is not partitioned the way you think, or the filter was applied inside a UDF the optimizer cannot see. Pushdown is a property of expressions the optimizer understands. A Python predicate hides the date from the prune.

Enterprise Real-World Scenarios

The inferred attempt number

A squad enabled the landing quickly by calling spark.read.json on the closed file with no schema, saving Parquet, and letting completion-v3 cast attempt_number to an integer in the warehouse. For three quiet weeks the sample inference chose integer, and the cast was a no-op. An exporter bug during a client release quoted attempt_number as a string in one afternoon's part of the file. Inference on a different sample the next day wrote a string column. The warehouse cast turned the new file's attempts into nulls inside ANSI-off SQL, and the completion definition, which ignores null attempts in a WHERE someone added later, dropped those learners. The landing job was green. The Monday number moved.

The repair was the declared PROGRESS_STRUCT, ANSI mode on the landing session, and the classify function's missing_or_bad_type reason. A fixture with a quoted attempt and a fixture with a boolean attempt are part of CI. Inference remains available in notebooks that never write the raw zone. The production entry point does not call json without .schema(...), and a review check greps for that. The incident was not a Spark defect. The reader did what an inferred schema is documented to do: guess from data. The contract was supposed to be the thing that does not guess.

A convenient Python timestamp

Another change parsed event_time with a Python UDF that called datetime.fromisoformat, because that was the function the lesson 14 job used and the author trusted it. The plan showed a Python evaluation on every row. A slice test on fifty thousand lines looked acceptable. The peak-day planning assumption, about fifty million lines, missed the remaining budget by a wide margin once the worker boundary and the single-threaded Python parse on each executor were in the profile. The JVM to_timestamp version of the same job, on the same cluster settings, finished inside the bound. The UDF had also used the driver's local display in a debug path that stripped the offset, so a subset of lines shifted by the zone of the executor image. Two executors in two zones would have made that worse. The session timezone was UTC. The UDF did not consult it.

The repair was the built-in parse and the rlike offset check in classify_lines. The Python standard library remains the right tool in the single-process job and in the driver's checkpoint write. It is the wrong per-row tool once the work is a DataFrame stage. The squad kept one fixture where Z and +00:00 must hash equally, so a future UDF that hashes the raw string fails CI. Performance and correctness failed together, which is common when the plan leaves the engine.

Best Practices

• Declare StructType on every production read. Keep inference in disposable notebooks.
• Treat nullable flags as documentation. Enforce the contract with explicit columns and tests.
• Turn ANSI mode on for the landing session and test bad casts.
• Read JSONL as text, inspect keys with json_object_keys, then from_json. Do not depend on the reader to drop unknown fields quietly.
• Use PERMISSIVE parsing with a visible reject reason. Do not use DROPMALFORMED on this file.
• Express time normalization, hashes, and filters with pyspark.sql.functions.
• Fail the run on more than one canonical hash per event_id. Collapse identical hashes only after that check.
• Keep actions few and small: counts, a one-row existence check, and the write. Do not show production rows.
• Read explain(mode="formatted") for the exchange and for column pruning before you scale the cluster.
• Put the collision rule in one function. A SQL view for reviewers should call that same DataFrame, not a forked query.
• Derive event_date in UTC inside the landing and carry event_time as the instant.
• Leave completion SQL out of these transformations.

Common Mistakes and Anti-Patterns

Schema inference on the landing. Types move with the sample, and the warehouse cast hides the damage. The repair is a frozen struct and fixtures for string, boolean, and missing attempts.

DROPMALFORMED or a filter that discards nulls with no count. Learners disappear and the job is green. The repair is a reject reason and a cap the driver enforces with an action.

Trusting the struct parser to reject email. Extra JSON fields are dropped when a schema is applied. The column never appears, and the file still published the rest. The repair is a key-set check before you trust the struct.

dropDuplicates as the identity policy. One of the conflicting payloads survives. The repair is countDistinct on the canonical hash and a refusal. dropDuplicates runs only to collapse exact retries.

A window that keeps the latest event_time. That is a winner policy. Producers can move the timestamp. The repair is the same refusal. Order is not a merge rule for this contract.

Python UDFs for row-wise contract logic. The stage leaves the JVM, the bound slips, and timezone handling forks from the engine. The repair is built-in functions, with a UDF reserved for a rule you cannot express and have measured.

collect of classified rows to decide in a Python for-loop. The driver becomes the lesson 14 process after you paid for the cluster, and payloads hit process memory. The repair is aggregations and limit(1) existence checks.

Two definitions, DataFrame and SQL, edited by different people. They drift, and only one is tested. The repair is one function. SQL is a view over it if a reviewer wants SQL.

Filter on event_date inside a UDF after a full scan. Partition pruning never sees the predicate. The repair is a column predicate in where on the partition column, confirmed in explain.

Troubleshooting and Design Review

Symptom: attempt_number arrives in Parquet as a string. The landing job succeeded.

Likely cause: the read inferred types, or a cast was skipped and the raw JSON string was written by a select of value.

Investigation: the read call, the schema passed in, and a fixture with a quoted number. Check the Parquet footer type. Do not guess from a dashboard.

Corrective action: declare the struct, backfill the affected dates from the closed input, and add the fixture. Do not cast in the warehouse and leave the publish drifting.

Symptom: A file containing email published, and the column is absent in the table.

Likely cause: schema-on-read dropped the field. The prohibited check never saw the key.

Investigation: json_object_keys on the raw line in a notebook that does not save results to a world-readable path. Confirm classify_lines is actually on the production path.

Corrective action: stop the publish, replace the date from a clean input, and put the key check in front of from_json. Assume the dropped field was still in the inbox file and handle that file under the raw-zone retention rule.

Symptom: The conflict job is much slower than the classify job, and shuffle bytes are close to the input size.

Likely cause: the aggregate includes full payloads or the raw line.

Investigation: explain the aggregate's expressions and the UI's shuffle write.

Corrective action: project to event_id and canonical before the group. Keep the wide stage narrow in record width.

Symptom: Duplicate event_id values with the same instant in two offset spellings remain two rows.

Likely cause: the hash covers the raw string, or to_timestamp failed one spelling and the rows were rejected unevenly, or dropDuplicates keys omitted the hash and also omitted a normalize step.

Investigation: the canonical struct in classify_lines, and the two fixture lines from this lesson.

Corrective action: hash the normalized UTC fields. Assert one accepted row for that fixture.

Symptom: explain shows a Python eval node and the stage misses the bound.

Likely cause: a UDF on the hot path.

Investigation: search the module for udf and pandas_udf. Read the plan.

Corrective action: replace with built-ins. Re-run the slice and compare stage time. Keep a UDF only with a written reason and a measurement.

Symptom: ANSI mode throws on a day that previously published.

Likely cause: bad casts were becoming nulls and being dropped. ANSI mode revealed them. The data did not get worse. The job got honest.

Investigation: the exception's column and a sample line in a private path.

Corrective action: quarantine or refuse according to the contract. Do not turn ANSI mode off to restore yesterday's green run. Yesterday's green run was dropping learners or coercing types.

A design review asks: Where is the schema literal? Which action refuses the file? Which action writes? What is the shuffle record? What does explain say about the exchange and about Python eval? How do identical retries collapse? What is the test fixture for email? Is completion computed here? The review can be a plan printout and a fixture run. It should be boring.

Interview Questions and Answers

Q1. What is a DataFrame, and what does a schema give you that an RDD of tuples does not?

A DataFrame is a distributed set of rows with named, typed columns. The optimizer can see those columns, drop the ones I do not use, and push predicates into the file scan. An RDD of tuples is opaque bytes plus a function. For progress events I want the schema so attempt_number cannot silently change type and so the collision aggregate shuffles a hash rather than a blob. The schema's nullability flags do not enforce themselves. I still check required fields.

Q2. Which operations in the landing are transformations, and which are actions?

classify_lines, the conflict grouping, and accepted_events are transformations. They return plans. count on the breach existence check, collect of the reject tally, and the Parquet write are actions. explain prints the plan and is not a data scan. I want two production actions when the file is clean: a small decision aggregate and a write. I do not want show. If I persist the narrow classified projection, I do it because two actions would otherwise parse twice, and I measure that, not by default.

Q3. Why is schema inference unsafe on this file?

Inference guesses from a sample. A quoted integer, a boolean, or a late column changes the guess, and the published Parquet type moves under completion-v3. I declare StructType, enable ANSI mode so bad casts fail, and test quoted, boolean, and missing attempts. Inference can stay in a notebook that does not write the zone.

Q4. Spark dropped the email field and the job succeeded. Where is the bug?

In the landing, for trusting a schema parse to enforce keys. With a declared schema, unknown JSON fields are omitted. I read text, call json_object_keys, and mark prohibited or unknown keys as a file breach before publish. The struct parse still applies types to the allowed fields. Both steps are required. Dropping email is not the same as refusing the file.

Q5. How do you collapse retries without hiding a payload conflict?

I normalize the contract fields, including UTC time, and sha2 a to_json struct with fixed field order. I countDistinct that hash per event_id. A distinct count above one fails the run. A distinct count of one may dropDuplicates on event_id and the hash so two identical lines publish once. dropDuplicates alone keeps an arbitrary payload. A window on event_time keeps a winner. Neither is the contract.

Q6. What should the shuffle carry for the collision check?

event_id and the canonical hash, and nothing else if I can help it. The group-by is the wide dependency and the stage boundary. If shuffle bytes approach the JSON size, the plan is carrying lines. I confirm with explain and the UI. I do not need learner_id in that aggregate. I need it in the publish, which can follow the narrow projection after the check passes.

Q7. When do you use Spark SQL instead of the DataFrame API?

When the expression is the contract a reviewer will read as a query, or when an analyst is pairing it with warehouse SQL they already know. It is the same Catalyst plan once the temporary view points at the classified DataFrame. I do not keep a second hand-written query that can drift. I do not saveAsTable the view into the warehouse and call it completion. Completion remains the downstream definition.

Q8. What does explain(mode="formatted") tell you that a green count does not?

Whether there is an exchange, which columns the scan reads, whether a partition filter was pruned, and whether a Python UDF sits on the row path. A green count on a fixture can hide a full-payload shuffle and a UDF that will miss the peak-day bound. I read the plan before I scale out. I still keep row-level assertions, because a plan string is not the business contract.

Q9. Why can limit(1) on a grouped conflict still read the whole file?

The aggregate is wide. Spark must see every copy of an event_id before it knows the distinct count. limit applies to the aggregated result. It is not a license to stop after the first line, except where a non-aggregate existence check can short-circuit. I budget a full pass for the collision job. I do not promise an early exit in the run book.

Q10. How do you keep event_date honest?

I reject timestamps without an offset, parse with built-ins under a UTC session zone, store event_time as UTC text, and derive event_date with date_format on that instant. The directory partition uses that column only. I filter null dates out of the write. I do not let each reader choose a zone. Two spellings of the same instant, Z and +00:00, become one hash in the fixture.

Q11. What is wrong with DROPMALFORMED on the progress file?

It discards lines and returns a green DataFrame with no remainder. A truncated tail or a partial export becomes a quiet hole in completion. I parse permissively, write a reason code for ordinary failures, and cap the count. File-level breaches never use that cap. Silence is not a mode I want on learner events.

Q12. Where does completion-v3 belong relative to these DataFrames?

Downstream of the published Parquet, in the SQL that defines completion, with the grain lesson 13 fixed. These DataFrames publish accepted events and keep checkpoint and event_type. They do not aggregate away attempts into a completion flag. If I add that aggregate here, I create a second definition the warehouse will eventually disagree with. The landing's SQL is the collision check, not the metric.

Hands-On Exercises

Exercise: Classify a fixture and read the plan

Problem statement. Implement classify_lines, conflict_keys, and accepted_events for progress JSONL. Prove the contract on small files. Do not write a production commit.

Tasks.

1. Create a local[2] session with UTC, ANSI mode, and a small shuffle partition count.
2. Build a fixture with an identical retry in two offset spellings, a malformed line, a missing field, a prohibited email, and a two-hash event_id.
3. Show that email is a file breach even though it is not a struct field.
4. Show that the two-hash id conflicts and that the identical retry does not.
5. Show one accepted row for the retry, with event_date 2026-03-02 and a Z instant.
6. Print explain for the conflict DataFrame and identify the exchange.
7. State which calls are actions, and what the driver collects.
8. Write what you would refuse to do with dropDuplicates alone, and where completion-v3 runs.

Expected deliverables. The functions, the fixture, the assertions, and a short plan note.

Evaluation rubric.

You pass when a reviewer can add a new prohibited field name to the fixture and predict a file breach without a publish, and can point at the exchange that enforces identity.

Lesson Summary

A DataFrame plan is the landing's description of the progress contract: a declared schema, a key check the schema parser will not do, built-in time normalization, and a shuffle that compares canonical hashes per event_id. Transformations build that plan. Actions that return counts decide. A write action publishes only after the decision. Spark SQL is the same optimizer when a reviewer wants a GROUP BY. Inference, DROPMALFORMED, Python UDFs on each row, and dropDuplicates as a conflict policy each produce a green job and a wrong mart. explain is how you see the exchange before enrollment week sees it for you.

PySpark Data Engineering wraps these functions in a job that writes a date partition, quarantines with a cap, and moves a checkpoint last.

Further Reading

• Spark SQL guide — DataFrames, temporary views, and the shared optimizer.
• JSON data source — modes, schema, and corrupt records. Pair it with json_object_keys in the SQL function reference so unknown keys are not a surprise.
• PySpark functions — from_json, to_timestamp, sha2, countDistinct. This is the API the classify function uses.
• ANSI SQL mode — what starts throwing once spark.sql.ansi.enabled is on. Confirm behavior on your patch level with the fixtures.
• Data Formats and Serialization — why the write, when you add it, is Parquet.
• Continue this course: PySpark Data Engineering, then Spark RDDs and Execution Model.