Spark: attest the committed dataset, not the scheduler stage.

Where Makoto sits

Executors never touch key material. The driver waits for the commit marker, then signs one file per artifact.

1. Driver startSigns the input file as an origin, keyed by the Spark application ID.
2. Writecoalesce(1) produces one part file so the output is one artifact.
3. Commit markerThe driver refuses to sign until _SUCCESS exists.
4. Transformation statementBinds the part file to the input bytes and origin statement.
5. GateVerifies the delivered bundle. A denial exits 1, so a scheduler wrapping spark-submit sees a failed job.

1. Install

Every platform on this site uses the same two files from the integration index: makoto_boundary.py, a standard-library helper that signs, hands off, and verifies through the pinned CLI, and makoto_bootstrap.py, which creates one key per claim role and the receiver’s trust policy. Put both next to the code below.

git clone https://github.com/makoto-project/makoto.git /opt/makoto
git -C /opt/makoto checkout 276008eeec0f10d5322c4c627b8c103222fee841
(cd /opt/makoto && uv sync --locked --dev)
export MAKOTO_CORE=/opt/makoto
uv pip install "pyspark==4.2.0"
export JAVA_HOME=/path/to/jdk-21

uv run makoto_bootstrap.py /srv/makoto \
  --source-kind urn:example:source:orders-export \
  --source-uri urn:example:orders:daily \
  --operation-type urn:example:operation:spark-normalize-orders
export MAKOTO_KEYS=/srv/makoto/keys MAKOTO_POLICY=/srv/makoto/policy.json

2. The job

Save this as orders_job.py.

"""A Spark job whose driver signs the committed output and gates the handoff.

    spark-submit orders_job.py <raw.csv> <workdir> [--tamper-in-transit]
"""

import shutil
import sys
import uuid
from pathlib import Path

from pyspark.sql import SparkSession
from pyspark.sql import functions as F

import makoto_boundary as mb

RECIPIENT = "example:analytics-team"


def main(raw: Path, work: Path, tamper: bool) -> int:
    app = SparkSession.builder.appName("makoto-orders").getOrCreate()
    run_id = app.sparkContext.applicationId  # stable for this application, unique per run
    run = work / run_id
    attestations = run / "attestations"
    event = lambda name: f"urn:uuid:{uuid.uuid5(uuid.NAMESPACE_URL, f'{run_id}/{name}')}"

    origin = mb.sign_origin(
        raw, attestations / "01-origin.dsse.json",
        source_kind="urn:example:source:orders-export",
        source_uri="urn:example:orders:daily",
        event_id=event("origin"),
    )

    orders = app.read.option("header", True).csv(str(raw))
    normalized = orders.withColumn("region", F.lower(F.trim("region"))).orderBy("order_id")
    output_dir = run / "orders_normalized"
    # One partition keeps one artifact. For many part files, attest a dataset
    # manifest that lists every partition instead.
    normalized.coalesce(1).write.mode("overwrite").option("header", True).csv(str(output_dir))
    app.stop()

    # Spark's committer writes _SUCCESS last; only then are the part files final.
    if not (output_dir / "_SUCCESS").exists():
        raise RuntimeError("output was not committed")
    (part,) = sorted(output_dir.glob("part-*.csv"))

    head = mb.sign_transform(
        part, attestations / "02-normalize.dsse.json",
        inputs=[(raw, origin)],
        operation_type="urn:example:operation:spark-normalize-orders",
        operation_name=f"makoto-orders {run_id}",
        event_id=event("normalize"),
    )
    anchors = mb.create_handoff(
        head, part, attestations, run / "bundle",
        recipient=RECIPIENT, bundle_id=event("bundle"), media_type="text/csv",
    )

    delivered = run / "receiver" / "bundle"
    shutil.copytree(run / "bundle", delivered)
    if tamper:
        final = next((delivered / "artifacts" / "final").glob("*.bin"))
        final.write_bytes(final.read_bytes() + b"1003,us,0.01\n")
    try:
        mb.verify(delivered, anchors, recipient=RECIPIENT, report_path=run / "receiver" / "verification.json")
    except mb.MakotoDenied as denied:
        print(f"REFUSED {run_id}: {denied}", file=sys.stderr)
        return 1
    print(f"ALLOWED {run_id}: {part.name}")
    return 0


if __name__ == "__main__":
    sys.exit(main(Path(sys.argv[1]).resolve(), Path(sys.argv[2]).resolve(), "--tamper-in-transit" in sys.argv))

3. Run it

spark-submit --master "local[2]" orders_job.py orders.raw.csv work

ALLOWED local-1789563868552: part-00000-973797d1-9009-4e71-9038-353f47389077-c000.csv

4. What a gate failure looks like

spark-submit --master "local[2]" orders_job.py orders.raw.csv work --tamper-in-transit
echo $?   # 1

REFUSED local-1789563873985: Makoto deny (E_ARTIFACT_DIGEST): step 12 E_ARTIFACT_DIGEST: final artifact bytes differ from the signed digest

Beyond one file

Many part files

Do not coalesce production data to one partition. Author a dataset manifest listing every part file and its digest, attest it with the core dataset-manifest profile, and hand off the partitions a receiver needs. The integration boundary notes describe the binding fields.

Delta Lake and Iceberg

Record the committed table version or snapshot ID as source metadata, and bind the exact data files, or a manifest over them, as the artifact. A version number alone is not bytes.

Structured Streaming

Close a bounded window in foreachBatch, write it, and sign that batch with the batch ID in the event ID. See the Kafka page for why the window itself must be checked too.

Other attachment points on this platform

Batch write

After the writer commits, build a dataset manifest over the exact output files and append one transformation statement.

Iceberg or Delta

Use the committed snapshot identifier as platform context; bind exact files or an authenticated manifest as the artifact.

Structured Streaming

Finalize a bounded offset/time window in foreachBatch, then attest that window as one artifact or dataset manifest.

Listener events

Capture application, job, stage, retry, and code context as operation metadata. Do not treat the event itself as the data.

What the evidence can say

It can bind a committed Spark output to signed predecessor statements and platform context. It cannot prove that every physical stage ran as reported, that Unity Catalog already provides Makoto evidence, or that a signature establishes data quality.

Troubleshooting

Could not find valid SPARK_HOME while searching

A pip-installed spark-submit must run with the virtual environment’s bin first on PATH, so its helper script finds pyspark.

ValueError: too many values to unpack at the part-file line

The output has more than one part file. Either coalesce, or switch to a dataset manifest as described above.