Use Invaract with Beam Python Pipelines

This guide checks an Apache Beam pipeline written with the Python SDK against a data contract before the pipeline is submitted, and shows how a platform team turns that on for a job whose source it does not own. The contract is the same YAML you write for Spark or for a Java Beam pipeline.

What this does and does not check

Invaract reads the shape of a Python pipeline: where it reads, where it writes, which columns and types come out, and in which format. It cannot read what your own Python functions do, so the transformation-shape rules (required_filter_columns, required_join_columns, required_group_by, forbid_cross_join) are refused with UNSUPPORTED_CONTRACT_FEATURE rather than checked. It recognises reads and writes of Parquet, Avro, text and TFRecord files, BigQuery tables and Pub/Sub topics. Static data-quality proofs and role consistency are refused the same way if you turn them on; fingerprints are not produced. The full list, in plain words, is in What Each Engine Checks.

How it works

Your job is unchanged. Two flags switch the check on:

• --runner=invaract_beam.InvaractRunner makes Beam hand the finished pipeline to Invaract’s runner, before any step has run.
• --invaract_delegate names the runner the pipeline goes to once it passes (DataflowRunner, FlinkRunner, …). The default is the DirectRunner.

The runner writes the pipeline’s portable form (the language-neutral description every Beam SDK produces) to a temporary file and runs Invaract’s checker on it. The checker is a Java program, the same code that checks Java pipelines, so Java 11 or later must be installed where the job is submitted. If the checker refuses the pipeline, or cannot run at all, the job does not start.

What a job needs

1. Install the invaract_beam package where the job is submitted. Until it is published, install it from the repository:

   pip install "git+https://github.com/mlltx/Invaract#subdirectory=beam-adapter/python"

2. Build the checker jar, and tell the package where it is:

   cd beam-adapter && sbt assembly
   export INVARACT_BEAM_CHECKER_JAR=$PWD/target/scala-2.12/invaract-beam-checker-0.1.0.jar

   --invaract_checker_jar=<path> on the job’s command line does the same.

3. Make sure the job builds its options from its command line, as Beam jobs usually do (PipelineOptions(argv)), and does not set its runner in code.

The adapter is tested with the Python SDK of Beam 2.68.0, 2.71.0, 2.74.0 and 2.77.0. A streaming pipeline that writes files needs 2.68.0 or later, the first release whose WriteToParquet takes triggering_frequency.

A pipeline that keeps to its contract

This contract is for a job that reads in/orders and writes out/report:

id: orders_report
version: "1.0.0"
inputs:
  - name: orders
    location: in/orders
    format: parquet
outputs:
  - name: report
    location: out/report
    format: parquet
    schema:
      fields:
        - { name: id, type: long, required: true }
        - { name: amount, type: long }

The job reads the orders, keeps the positive ones and writes a report:

SCHEMA = pa.schema([pa.field("id", pa.int64(), nullable=False), ("amount", pa.int64())])

with beam.Pipeline(options=PipelineOptions(beam_args)) as p:
    (p | "ReadOrders" >> beam.io.ReadFromParquet(f"{root}/in/orders/*.parquet")
       | "Positive" >> beam.Filter(lambda r: (r["amount"] or 0) > 0)
       | "WriteReport" >> beam.io.WriteToParquet(f"{root}/{output}", SCHEMA, file_name_suffix=".parquet"))

A platform runs it with the runner and the contract:

python example_job.py --root=. \
  --runner=invaract_beam.InvaractRunner \
  --invaract_contract=contracts/orders_report.yaml

The pipeline is checked, passes, and runs on the DirectRunner as before, writing out/report-00000-of-00001.parquet. With --invaract_notify_config a PASSED validation event naming the engine beam-python is published (see Notification Sinks).

A pipeline that does not

If the same job writes somewhere else (--output=out/report_v2), the job stops before anything runs, with the reason (the job’s own traceback is left out):

invaract_beam.runner.ContractViolation: Contract violation: 'orders_report@1.0.0' rejected this transformation. Write aborted.

What the contract expects:
  input  'orders' at in/orders:
  output 'report' at out/report: id: long, amount: long (optional)

What the plan contains:
  Write(./out/report_v2, format=parquet)
  └─ UnknownPlan(Positive (ParDo))
     └─ Read(./in/orders)

Why it violates the contract (1 violation):
  1. [OUTPUT_LOCATION_MISMATCH] contract declares output location 'out/report' but the plan writes to './out/report_v2'

How to correct it:
  1. Write to 'out/report' instead, or update the contract's declared output location to './out/report_v2' if this location change is intentional.

The exception is invaract_beam.ContractViolation. If the check could not be made at all (no Java, no checker jar, a setting the checker does not know), the exception is invaract_beam.CheckNotRun, and the job does not start either.

Under “What the plan contains”, a collection that feeds more than one step (a branch that is merged again later, or used on both sides of a join) is drawn in full, with the steps that produce it, the first time. Each later place it feeds is a single line ending (shown above), so the explanation stays short however many times a pipeline splits and rejoins. A read on its own is one line, so it is simply repeated.

The settings

Every Invaract setting is a flag of the job, spelled --invaract_<setting> (all values are text, so a switch is --invaract_dry_run=true):

Flag	What it does
--invaract_contract	The contract to enforce: a file path, or registry://<id>@<version> (Use a Contract Registry).
--invaract_dry_run=true	Infer a contract from the job’s writes instead of enforcing one (Dry-Run Mode).
--invaract_notify_config	Where validation events go (Notification Sinks).
--invaract_location_map	How ref:// locations resolve (Resolve Dataset Locations).
--invaract_org_policy, --invaract_org_policy_overlays	Organizational policy layered on the contract (Enforce Organizational Policy).
--invaract_reject_undeclared_inputs=true, --invaract_reject_undeclared_fields=true	Stricter checking of reads and columns.
--invaract_trusted_steps	Steps that end the pipeline without a write that a platform trusts (see below).
--invaract_registry_url, --invaract_registry_client_class	The contract registry.

And how the checker is run:

Flag	What it does
--invaract_delegate	The runner a passing pipeline is handed to. Default: DirectRunner.
--invaract_check_only=true	Check the pipeline and hand it to no runner, for a CI build (see below).
--invaract_checker_jar	The checker jar. Default: the INVARACT_BEAM_CHECKER_JAR environment variable.
--invaract_java	The java program. Default: $JAVA_HOME/bin/java, else java on the PATH.
--invaract_java_options	Options for the checker’s JVM, such as -Xmx512m.

Where locations come from

• A file read is located at the folder its pattern reads: data/in/orders/*.parquet reads data/in/orders.
• A file write is located by its pattern up to the shard template: WriteToParquet("data/out/report") writes data/out/report-00000-of-00001.parquet, located at data/out/report.
• A BigQuery table project:dataset.table is project/dataset/table, registered in the bigquery catalog.
• A Pub/Sub topic or subscription is its full path (projects/p/topics/t).

Formats, types and what cannot be told

• Columns are known when the write states the schema it writes (a Parquet write’s pyarrow schema, an Avro write’s schema, a BigQuery load’s table schema) or when the data is typed Beam rows (ReadFromParquet(..., as_rows=True), a beam.Map(...).with_output_types(...) that yields rows). A write of plain Python dictionaries states no columns: if its output declares columns, the write is refused rather than passed unchecked. A beam.Select with a computed field has an untyped column, which counts as unknown.
• Formats are the sink a write uses. A text write cannot tell CSV from JSON.
• Save modes: only a BigQuery write states one (WRITE_APPEND, WRITE_TRUNCATE, WRITE_EMPTY are append, overwrite, error). A contract that declares a save mode for a file write has that write refused.
• Reads that state no location are still seen, never mistaken for data the job makes: a CSV read through pandas (ReadFromCsv), a BigQuery query, a source class Invaract does not know, files matched by fileio (MatchFiles, MatchContinuously), a classic template’s parameter. A declared input such a read may be is reported unverifiable rather than missing, and with --invaract_reject_undeclared_inputs=true the write is refused.
• Writes Invaract cannot place are refused: a CSV write through pandas (WriteToCsv), fileio.WriteToFiles, a BigQuery write whose table the job computes, a classic template’s parameter.

Steps that end the pipeline

A step that ends the pipeline without being a write Invaract recognises (a beam.Map(print), a call to a service) is refused, because it may move data where the contract cannot see. Two kinds are let through, exactly as for Java pipelines: a step that reads only what a write reports (the file names a write produced), and a step a platform names as trusted: --invaract_trusted_steps=Notify*,Metrics.

The refusal names the step and says which flag would trust it. For a job that ends with rows | "PushToService" >> beam.Map(print):

Why it violates the contract (1 violation):
  1. [UNVERIFIABLE_WRITE] A step that ends the pipeline but is not a recognised write: PushToService (ParDo). It was never checked against contract 'orders_report@1.0.0'.
How to correct it:
  1. If this step moves no data the contract governs (it logs, counts or notifies), a platform can name it as trusted with --invaract_trusted_steps=PushToService. If it writes data, write through an IO transform Invaract recognises, with its destination fixed when the pipeline is built - see "Steps that end the pipeline" in the docs' guide (guides/using-invaract-with-beam-python).

A write Invaract cannot place (a destination the job computes, a template parameter) is refused the same way, but naming it as trusted does not let it through, and its remediation says so.

Dry-run mode

With --invaract_dry_run=true and no contract, the job runs unchanged and Invaract writes the contract each write would need. It is the usual way to start on an existing job:

python example_job.py --root=. \
  --runner=invaract_beam.InvaractRunner \
  --invaract_dry_run=true \
  --invaract_notify_config=notify.properties \
  --job_name=orders-report

With a notification sink configured, each write’s draft is published as a CONTRACT_INFERENCE event. For the example job above, its contractYaml is (real output, Beam 2.77.0):

id: inferred_contract
version: 0.1.0
status: draft
inputs:
- name: input
  location: ./in/orders
outputs:
- name: output
  location: ./out/report
  format: parquet
  schema:
    fields:
    - name: id
      type: long
      required: true
      nullable: false
    - name: amount
      type: long
      required: true
      nullable: true
  derivedFrom:
  - input

The output’s columns come from the schema WriteToParquet is given. The input has none, because a ReadFromParquet of dictionaries leaves its file schema for Python to read when the job runs. The event says so, with status INFERRED_DEGRADED and the reason the pipeline states no columns for the read of ./in/orders, so the draft declares none for it. Read with as_rows=True, or fill the columns in yourself when you review the draft.

After the last write, one DRY_RUN_SUMMARY event merges every draft into one contract named after --job_name (here id: orders-report, with the input named orders and the output report). It is published before the pipeline runs, since the whole pipeline is known by then. Add --invaract_check_only=true to get the same events without running the job at all. With no sink, the drafts and the summary are logged instead. See Dry-run mode for how to turn a draft into a contract.

Check a pipeline in CI, without running it

Add --invaract_check_only=true to the job’s production command line and run the job in a CI build:

python -m my_job \
  --runner=invaract_beam.InvaractRunner \
  --invaract_contract=contracts/orders_report.yaml \
  --invaract_delegate=DataflowRunner --project=my-project --region=europe-west1 \
  --invaract_check_only=true

The job builds its pipeline as always and the checker checks it, but the pipeline is then handed to no runner: run() returns invaract_beam.CheckedOnly, a finished result (state DONE, no metrics), so a job that waits for its pipeline ends normally. Nothing is read or written, and the runner --invaract_delegate names is not created. A pipeline that breaks its contract raises ContractViolation exactly as it would in production, so the build fails with the same explanation. With --invaract_dry_run=true in place of a contract, the same command reports the contract the job would need, without running it. Events from a check-only run have "checkOnly": "true" in their job.engineDetails, so a platform can tell a CI check from a production run.

What the build needs: Java and the checker jar, as any check does, and the job’s own dependencies, but no cluster or credentials. Python’s BigQuery and Pub/Sub transforms contact nothing while the pipeline is built (confirmed with Beam 2.77). File reads are the exception, and it is Beam’s own: they check that their pattern matches some file when the pipeline is built (confirmed for ReadFromText), so give the build the input files or build the read with validate=False.

When the command line cannot reach the job

A job that sets its runner in code (PipelineOptions(argv, runner="DirectRunner")) or does not pass its command line to Beam is not checked: the flags never reach it. A job submitted from a machine without the package does not start (ModuleNotFoundError: No module named 'invaract_beam'), and one submitted without Java stops with CheckNotRun.

Related

• Configuration Settings: every --invaract_... flag, with its Spark and Java spelling.
• What Each Engine Checks: what each engine checks and what it cannot, in plain words.
• Engine Capabilities: the same, per capability, with the adapters’ own notes.
• Connector Support: the IOs the checker recognizes.
• Common Problems: what each CheckNotRun message means.