Module 9 — Writing results and downstream integration

A Spark job that computes without writing anything is a demo. The last piece of engineering is turning the trained pipeline and the scored rows into artifacts that another job — or another team — can read. This module covers three things: how to write DataFrames well, how to persist and reload a PipelineModel, and how a scheduled scoring job is wired.

Writing a DataFrame — partitioned by design

The default write is a single call whose defaults you should override on purpose every time:

(
    predictions
    .select("flight_id", "carrier", "year", "month", "prediction", "probability")
    .write
    .mode("overwrite")
    .partitionBy("year", "month")
    .parquet("s3://predictions/flights/")
)

Three decisions in this snippet deserve to be spelled out.

partitionBy("year", "month") lays the output on disk exactly the way module 4 described the input. Downstream jobs that only need a month of predictions read that one folder, and partition pruning does the rest. Pick partition columns with the same rules as module 4 — low cardinality, always in the WHERE clause of downstream queries.

.select(...) before .write trims the output to the columns downstream consumers actually need. Spark is happy to write two hundred columns, and every one of them costs disk space and read time.

The write mode is not optional — pick it consciously, because it is what most often breaks a scheduled job.

The four write modes

Mode	Behavior	When to use
error (default)	Fail if the target path exists	Never in production; the default is bait
overwrite	Delete the target, then write	Batch retrains; one-off analyses
append	Add new files to the existing target	Streaming, incremental daily loads
ignore	Silently do nothing if target exists	Idempotent bootstrap of a shared table

Two subtleties save real production incidents. First, overwrite on a partitioned table by default deletes every partition — even those your current DataFrame does not include. To restrict the overwrite to only the partitions you are writing, set spark.sql.sources.partitionOverwriteMode=dynamic. This turns "overwrite the January 2024 partition and leave the rest alone" into the natural behavior.

Second, append on Parquet is not transactional. A failure halfway through leaves half of the new files behind, and there is no rollback. For truly transactional appends, look at Delta Lake or Iceberg — both plug into the same DataFrame writer and add ACID semantics on top of Parquet.

Saving the trained pipeline

The PipelineModel from module 6 knows how to write itself. It is a plain folder of parameters and small binary blobs; there is nothing to lose in translation.

pipeline_model.write().overwrite().save("s3://models/flights-delay-gbt-v1/")

Version the folder name, not the file inside it. flights-delay-gbt-v1, flights-delay-gbt-v2, and never mutate a version in place. Rollbacks then become a single-parameter change in the scoring job.

Reloading is symmetric and does not require the training code:

from pyspark.ml import PipelineModel

model = PipelineModel.load("s3://models/flights-delay-gbt-v1/")
scored = model.transform(new_flights)

The reloaded pipeline includes the fitted string-indexer vocabularies, the scaler statistics, and every tree of the gradient-boosted classifier. There is no separate scaler file to remember to bundle — the discipline of module 6 pays off here.

A scheduled scoring job, from end to end

Batch scoring on the flight-delays project runs once a day and looks like this in production:

from datetime import date

def score_day(spark, day: date):
    input_path  = "s3://flights-parquet/"
    output_path = "s3://predictions/flights/"
    model_path  = "s3://models/flights-delay-gbt-v1/"

    flights = (
        spark.read
        .parquet(input_path)
        .filter((F.col("year") == day.year) & (F.col("month") == day.month) & (F.col("day") == day.day))
    )

    model = PipelineModel.load(model_path)
    scored = model.transform(flights).select(
        "flight_id", "carrier", "year", "month", "day", "prediction", "probability"
    )

    (
        scored
        .coalesce(4)                          # module 8: fewer, bigger output files
        .write
        .mode("overwrite")
        .partitionBy("year", "month", "day")
        .parquet(output_path)
    )

Every element of this function comes from an earlier module: partitioned read with pruning (4), pipeline reload (6), coalesce before write (8), partitioned output (this module). The job is scheduled — Airflow, Databricks Workflows, a cron on a small cluster — and its idempotency comes entirely from the dynamic partition overwrite plus the day-scoped filter.

What deploys and what does not

The model is a folder in object storage. The scoring code is a Python file in the repository. Update either independently: swap the model folder to roll a new version, edit the code to fix a bug in the schema, redeploy. Keeping these two axes independent is the payoff of everything this course has done.

Summary

• Write Parquet, partitionBy low-cardinality columns, .select() before .write to trim.
• Pick a write mode consciously; overwrite deletes every partition unless partitionOverwriteMode=dynamic.
• Save PipelineModel as a versioned folder; never mutate a version in place; reload symmetrically with .load().
• Scheduled scoring is a small function that combines partitioned reads, pipeline reload and partitioned writes — no ad hoc plumbing.

Next module: the full project on tens of millions of rows, with a pandas control on a sample of the same data.