Spark DataFrames and Spark SQL

Writing a result Spark can inspect

A daily revenue query must decide which orders qualify, how customers match, and which calendar day receives each amount. Spark can optimize the computation only after those choices are expressed. The optimizer does not decide whether a refund belongs in revenue or whether a missing customer should remove an order.

A Spark DataFrame represents rows with a schema and a relational computation. select chooses columns or expressions, filter keeps rows whose condition is true, join combines matching rows, and groupBy with an aggregation changes the output grain. These operations normally build a plan; actions such as collect or a write request execution. Schema inference, file discovery, and analysis can still perform work before the final action.

In PySpark, F.col(“amount_cents”) > 0 creates a Column expression, not a Python boolean. Combine column conditions with parenthesized & and | expressions rather than Python and/or; use isNull for null tests. A filter retains true rows and discards false or null conditions. Give derived columns explicit aliases so later SQL and tests use stable names.

The examples assume familiarity with Python functions and basic SQL. They use small local Spark datasets to connect DataFrame expressions with SQL, then test parsing, joins, numeric limits, reporting days, and Python functions. This is batch processing; it does not cover streaming state or production performance tuning.

From logical intent to physical operators

A logical plan describes operations such as filtering and grouping without committing to one execution algorithm. Spark resolves names and types, applies valid Catalyst rewrites, and chooses a physical plan with scans, joins, aggregations, and exchanges. Physical operators and scheduler stages are related but are not the same objects. explain(“extended”) exposes parsed, analyzed, optimized, and physical plans.

For a country-level sum, column pruning can remove an unused note, but country remains necessary for filtering and grouping even if another query omits it from the final output. Predicate movement through a join must preserve join and null semantics. A right-side filter in WHERE after a left join can remove unmatched rows; putting the same filter in ON or filtering the right input can preserve them. These forms are not interchangeable.

Predicate pushdown into a file reader is a further, source-dependent step. Supported predicates and usable statistics may reduce work, but a residual Filter can remain. Inspect ReadSchema and PushedFilters for a Parquet scan. The presence of a pushed predicate alone does not establish bytes skipped or prove that an entire filter was evaluated by the reader.

The DataFrame API and SQL share Spark SQL planning. Equivalent queries often receive similar plans, but equal results do not guarantee identical plan text or identical execution across versions, settings, hints, and source statistics. Compare the full result and schema first. Use the plan to understand execution choices rather than as a universal equality test. A temporary view provides a SQL name for a DataFrame in a session; creating it does not materialize the rows.

A schema needs a rejection policy

A schema declares structure, not every business rule. Inference is useful for exploration and can be appropriate for controlled sources; it is not inherently incorrect. CSV inferSchema makes an extra pass and its result depends on the values and options. An amount containing N/A can become a string. Spark SUM does not concatenate strings: supported casts may occur, or invalid input can fail or become null depending on the operation and settings.

An explicit schema alone does not quarantine records. CSV PERMISSIVE parsing can place invalid fields at null; retaining corrupt record text requires the appropriate field and options, and detection can depend on the columns parsed. FAILFAST rejects parsing failures. Neither mode enforces uniqueness, valid business ranges, or every required-field rule. Check the relevant reader contract rather than assuming nullable=False is a complete validation gate.

The lab keeps amount text, uses try_cast, and labels missing and invalid amounts separately. Accepted and rejected row counts must account for all input rows. This creates inspectable rejection DataFrames, not a durable quarantine service. Production also needs source identity, retained raw input, reasons, a publication policy for partial results, and a correction path. CSV structural errors need their own reader policy; this fixture tests a valid CSV shape with bad field values.

Join matches must be counted per input row

For an inner equality join, a left row with m matching right rows produces m output rows; with no matches it disappears. A left join produces max(1, m) rows. Right-key uniqueness therefore preserves the left-join count, but an inner join also needs complete matching. Equal total counts alone can hide a dropped row offset by a duplicated row.

Check duplicate keys, unmatched input identities, and match counts per input row, then reconcile comparable totals. Use a known non-null right-side marker to count matches; a nullable descriptive attribute can be null even on a matched row. These checks can be expensive at scale, so place them at the appropriate materialization boundary and reuse the evidence.

With ordinary equality, null does not match null. Spark’s null-safe equality, written <=> in SQL or eqNullSafe in PySpark, explicitly changes that rule. It can also multiply null-key rows if both sides contain several of them. An SCD table needs a deliberate current-version or point-in-time condition; arbitrarily dropping duplicate keys does not choose the correct historical record.

Represent money and reporting days explicitly

Use a currency and scale with integer minor units or suitable decimal values. Decimal(p, s) has p total digits and s fractional digits; decimal(5,2) can hold 999.99 but not 1000.00. Arithmetic and casts can change scale, round, or overflow, and integer types also have limits. Spark caps decimal precision at 38. Set rounding and overflow policies and build decimal inputs from source decimal strings rather than binary floats.

The lab checks an exact 0.30 decimal sum, a scale-reducing cast of 1.235 to 1.24, and overflow detected by try_cast returning null. This does not claim decimal arithmetic is free or exact at every intermediate scale. Floating-point accumulation error depends on values, order, and algorithm; a tiny example does not justify a fixed monetary drift at a billion rows.

Spark TIMESTAMP_LTZ represents an instant whose calendar interpretation uses the session time zone; it does not retain each input’s original zone name. TIMESTAMP_NTZ represents local date-time fields without a time zone and is not, by itself, a unique instant. Record which meaning the source supplies. Naive local times need an explicit zone and an ambiguity policy at daylight-saving transitions.

The examples set the session to UTC, parse offset-bearing strings as instants, then use from_utc_timestamp with Asia/Seoul before extracting the report date. This conversion pattern relies on the UTC session. Do not shift an already shifted local value again. The two instants around UTC midnight fall on different UTC dates and the same Seoul date. A named zone handles its calendar rules; a fixed numeric offset is not a substitute for regions whose offset changes.

A Python UDF is a boundary to inspect

Built-in expressions expose their structure and types to Spark SQL. A Python UDF supplies a callable with a declared return type whose internals are generally opaque to Catalyst. Scalar Python execution crosses a Python worker boundary; Arrow-enabled execution changes transfer and batching, and pandas UDFs use batch-oriented APIs. These distinctions do not imply a universal runtime ratio or that every built-in uses the same columnar execution path.

A UDF is not required to run last. Spark may prune unused results and move eligible work around it, but it cannot generally translate its arbitrary Python body into a source predicate. A surrounding conditional is not a guarantee that an unsafe UDF will never see null or invalid input; make the function safe on its documented input domain.

The example compares a null-safe floor-division UDF with a built-in expression for bounded integer cents. For -101, Python // 100 is -2, while a division followed by truncation toward zero would give -1. The original positive-only comparison missed this difference. Whole units here deliberately discard cents; this is not an exact money conversion. The lab checks signed and null results and Python evaluation nodes, not a tenfold speed claim. Benchmark equivalent computations separately with repeated runs, consistent inputs, and controlled cache and warm-up conditions.

Test a zero-only day, not just a zero row

For the daily result in this article, a day with a paid zero-amount order is present with zero revenue; days with no paid orders are absent. This is a reporting choice. Refund-status rows are excluded in this fixture, not a universal accounting rule. A ledger that records negative refund movements needs a different definition.

Place the zero order on its own day. If it shares a day with a positive order, deleting the zero row leaves the result unchanged and cannot test day preservation. Compare the complete sorted output, including the zero-only day. Checking uniqueness after converting to a Python dictionary is too late because duplicate keys may already have been collapsed; inspect the DataFrame rows or group counts first.

The final exercise tests midnight separation, exclusion of the refund row, the zero-only day, and empty input. It also shows that removing zero rows breaks the expected answer. Production tests should add null amounts and timestamps, duplicate event identities, currency handling, overflow, and the chosen local-calendar boundaries. Deterministic fixtures and representative production checks serve different purposes; neither replaces the other.

Lab: actual local Spark execution

Use Python 3.12, pyspark==4.2.0, and Java 17 with JAVA_HOME set to that installation. Driver and Python workers use the same Python minor version. Run the setup, then each example in a standalone process without another Spark session. local[2] uses local worker threads and loopback communication, not a multi-node cluster. The helper stops Spark and removes its temporary files. SQL settings explicitly fix UTC, TIMESTAMP_LTZ, ANSI mode, and two shuffle partitions; AQE is disabled to keep the inspected plan simple. Spark startup warnings about logging, incubator modules, and the native Hadoop fallback are separate from recorded stdout. No runtime speedup or large-data memory claim is tested.

import os, sys, tempfile
from pathlib import Path
from contextlib import contextmanager, redirect_stdout
from io import StringIO
from decimal import Decimal
from pyspark.sql import SparkSession, functions as F, types as T

os.environ["PYSPARK_PYTHON"] = sys.executable
os.environ["SPARK_LOCAL_IP"] = "127.0.0.1"

@contextmanager
def local_spark():
    with tempfile.TemporaryDirectory(prefix="de021-") as work:
        spark = (SparkSession.builder.master("local[2]").appName("dataframe-lab")
                 .config("spark.driver.host", "127.0.0.1")
                 .config("spark.driver.bindAddress", "127.0.0.1")
                 .config("spark.ui.enabled", "false")
                 .config("spark.ui.showConsoleProgress", "false")
                 .config("spark.sql.warehouse.dir", work)
                 .config("spark.sql.session.timeZone", "UTC")
                 .config("spark.sql.timestampType", "TIMESTAMP_LTZ")
                 .config("spark.sql.ansi.enabled", "true")
                 .config("spark.sql.shuffle.partitions", "2")
                 .config("spark.sql.adaptive.enabled", "false")
                 .getOrCreate())
        spark.sparkContext.setLogLevel("ERROR")
        try:
            yield spark, Path(work)
        finally:
            spark.stop()

def explain_text(df):
    out = StringIO()
    with redirect_stdout(out):
        df.explain(mode="extended")
    return out.getvalue()

def sorted_rows(df):
    return sorted(tuple(r) for r in df.collect())

1. Compare DataFrame and SQL results and inspect the Parquet scan.

with local_spark() as (spark, work):
    source = spark.createDataFrame(
        [(1, "KR", 100, "unused"), (2, "US", 300, "unused"), (3, "KR", 200, "unused")],
        "id long, country string, amount_cents long, note string")
    path = str(work / "orders")
    source.write.parquet(path)
    orders = spark.read.parquet(path)
    orders.createOrReplaceTempView("orders")
    api = (orders.filter(F.col("country") == "KR")
           .groupBy("country").agg(F.sum("amount_cents").alias("revenue")))
    sql = spark.sql("SELECT country, SUM(amount_cents) AS revenue FROM orders WHERE country = 'KR' GROUP BY country")
    assert sorted_rows(api) == sorted_rows(sql) == [("KR", 300)]
    plan = explain_text(api)
    scan = next(line.strip() for line in plan.splitlines() if "ReadSchema:" in line)
    read_schema = scan.split("ReadSchema:", 1)[1]
    assert "country:string" in read_schema and "amount_cents:bigint" in read_schema
    assert "note:" not in read_schema and "id:" not in read_schema
    print("API and SQL rows:", sorted_rows(api))
    print("scan retains filter and amount columns:", "country:string" in read_schema)
    print("unused columns pruned:", "note:" not in read_schema and "id:" not in read_schema)
    print("country predicate offered to reader:", "EqualTo(country,KR)" in scan)
# API and SQL rows: [('KR', 300)]
# scan retains filter and amount columns: True
# unused columns pruned: True
# country predicate offered to reader: True

2. Separate invalid amounts from missing values and account for every input row.

with local_spark() as (spark, work):
    path = work / "input.csv"
    path.write_text("id,amount_cents\n1,6300\n2,N/A\n3,\n4,4200\n", encoding="utf-8")
    inferred = spark.read.option("header", True).option("inferSchema", True).csv(str(path))
    raw = spark.read.option("header", True).schema("id string, amount_cents string").csv(str(path))
    typed = raw.select("*", F.expr("try_cast(amount_cents AS BIGINT)").alias("amount"))
    checked = typed.withColumn("reason", F.when(F.col("amount_cents").isNull(), "missing_amount")
                               .when(F.col("amount").isNull(), "invalid_amount"))
    accepted = checked.filter(F.col("reason").isNull())
    rejected = checked.filter(F.col("reason").isNotNull())
    assert sorted_rows(accepted.select("id", "amount")) == [("1", 6300), ("4", 4200)]
    assert sorted_rows(rejected.select("id", "reason")) == [("2", "invalid_amount"), ("3", "missing_amount")]
    assert accepted.count() + rejected.count() == raw.count() == 4
    print("inferred amount type:", inferred.schema["amount_cents"].dataType.simpleString())
    print("accepted:", sorted_rows(accepted.select("id", "amount")))
    print("rejected:", sorted_rows(rejected.select("id", "reason")))
    print("all source rows accounted for:", accepted.count() + rejected.count() == 4)
# inferred amount type: string
# accepted: [('1', 6300), ('4', 4200)]
# rejected: [('2', 'invalid_amount'), ('3', 'missing_amount')]
# all source rows accounted for: True

3. Count duplicate, unmatched, and null-key join results.

with local_spark() as (spark, work):
    orders = spark.createDataFrame([(1, "C1", 100), (2, "C2", 200), (3, None, 300)],
                                   "order_id long, customer_id string, amount long")
    customers = spark.createDataFrame([("C1", "basic"), ("C1", "premium"), (None, "unknown")],
                                      "customer_id string, segment string")
    inner = orders.join(customers, "customer_id", "inner")
    left = orders.join(customers, "customer_id", "left")
    matches = left.groupBy("order_id").agg(F.count("segment").alias("matches"))
    assert sorted_rows(matches) == [(1, 2), (2, 0), (3, 0)]
    unique = customers.filter(F.col("customer_id").isNotNull()).dropDuplicates(["customer_id"])
    assert orders.join(unique, "customer_id", "left").count() == 3
    assert orders.join(unique, "customer_id", "inner").count() == 1
    nullsafe = orders.alias("o").join(customers.alias("c"),
        F.col("o.customer_id").eqNullSafe(F.col("c.customer_id")), "inner")
    print("source, inner, left rows:", orders.count(), inner.count(), left.count())
    print("matches by order:", sorted_rows(matches))
    print("null-safe inner rows:", nullsafe.count())
    assert nullsafe.count() == 3
# source, inner, left rows: 3 2 4
# matches by order: [(1, 2), (2, 0), (3, 0)]
# null-safe inner rows: 3

4. Test decimal limits and two reporting calendars.

with local_spark() as (spark, work):
    amounts = spark.createDataFrame([("0.10",), ("0.10",), ("0.10",)], "raw string")
    total = amounts.select(F.col("raw").cast("decimal(18,2)").alias("amount")).agg(F.sum("amount")).first()[0]
    assert total == Decimal("0.30")
    limits = spark.sql("SELECT try_cast('1000.00' AS DECIMAL(5,2)) AS overflow, CAST('1.235' AS DECIMAL(5,2)) AS rounded").first()
    assert limits.overflow is None and limits.rounded == Decimal("1.24")
    times = spark.createDataFrame([("2026-03-01T23:59:00Z",), ("2026-03-02T00:01:00Z",)], "raw string")
    days = times.select(F.to_timestamp("raw").alias("instant")).select(
        F.to_date("instant").cast("string").alias("utc_day"),
        F.to_date(F.from_utc_timestamp("instant", "Asia/Seoul")).cast("string").alias("seoul_day"))
    assert sorted_rows(days) == [("2026-03-01", "2026-03-02"), ("2026-03-02", "2026-03-02")]
    print("decimal sum:", total)
    print("overflow detected, rounded:", limits.overflow is None, limits.rounded)
    print("UTC and Seoul days:", sorted_rows(days))
# decimal sum: 0.30
# overflow detected, rounded: True 1.24
# UTC and Seoul days: [('2026-03-01', '2026-03-02'), ('2026-03-02', '2026-03-02')]

5. Compare UDF and built-in semantics for signed and null input.

with local_spark() as (spark, work):
    values = spark.createDataFrame([(1, -101), (2, -1), (3, 0), (4, 101), (5, None)], "id long, cents long")
    @F.udf(T.LongType(), useArrow=False)
    def whole_units(cents):
        return None if cents is None else cents // 100
    via_udf = values.select("id", whole_units("cents").alias("whole"))
    via_builtin = values.select("id", F.floor(F.col("cents").cast("decimal(20,0)") / F.lit(100)).cast("long").alias("whole"))
    expected = [(1, -2), (2, -1), (3, 0), (4, 1), (5, None)]
    assert sorted_rows(via_udf) == sorted_rows(via_builtin) == expected
    udf_plan, builtin_plan = explain_text(via_udf), explain_text(via_builtin)
    assert "BatchEvalPython" in udf_plan and "BatchEvalPython" not in builtin_plan
    print("same signed and null results:", sorted_rows(via_builtin))
    print("Python evaluation node:", "BatchEvalPython" in udf_plan)
    print("built-in avoids Python evaluation:", "BatchEvalPython" not in builtin_plan)
# same signed and null results: [(1, -2), (2, -1), (3, 0), (4, 1), (5, None)]
# Python evaluation node: True
# built-in avoids Python evaluation: True

6. Preserve a zero-only day and test the empty input.

def daily_revenue(df, zone="UTC"):
    return (df.filter(F.col("status") == "paid")
            .withColumn("day", F.to_date(F.from_utc_timestamp("instant", zone)))
            .groupBy("day").agg(F.sum("amount_cents").alias("revenue_cents")))

with local_spark() as (spark, work):
    fixture = spark.createDataFrame([
        (1, "paid", 1000, "2026-03-01T23:59:00Z"),
        (2, "paid", 500, "2026-03-02T00:01:00Z"),
        (3, "refunded", 700, "2026-03-02T12:00:00Z"),
        (4, "paid", 0, "2026-03-03T12:00:00Z")],
        "id long, status string, amount_cents long, raw_time string")
    fixture = fixture.withColumn("instant", F.to_timestamp("raw_time"))
    out = daily_revenue(fixture)
    actual = sorted_rows(out.select(F.col("day").cast("string"), "revenue_cents"))
    expected = [("2026-03-01", 1000), ("2026-03-02", 500), ("2026-03-03", 0)]
    assert actual == expected
    assert out.groupBy("day").count().filter(F.col("count") != 1).count() == 0
    empty = daily_revenue(fixture.limit(0))
    assert empty.count() == 0
    zero_removed = daily_revenue(fixture.filter(F.col("amount_cents") != 0))
    assert sorted_rows(zero_removed.select(F.col("day").cast("string"), "revenue_cents")) != expected
    print("daily rows:", actual)
    print("empty input has no invented day:", empty.count() == 0)
    print("fixture catches dropping zero-only day:", zero_removed.count() == 2)
# daily rows: [('2026-03-01', 1000), ('2026-03-02', 500), ('2026-03-03', 0)]
# empty input has no invented day: True
# fixture catches dropping zero-only day: True

References: Spark SQL, CSV reader options, Null semantics, ANSI and type conversion, Datetime types, Python UDFs.