Install the Enforcement Rule on Spark

This guide wires Invaract into a Spark job of your own. (For Apache Beam, see Use Invaract with Apache Beam or Beam Python pipelines.) ContractEnforcementRule is a SparkSessionExtensions check rule: it runs on the analyzed logical plan of every query your session executes, and throws before Spark runs a query that violates the active contract.

Attach it via `spark-submit --conf`

InvaractSparkSessionExtension installs Invaract with no code at all in the job it’s attached to — the same way Delta Lake’s own io.delta.sql.DeltaSparkSessionExtension is attached, purely via spark.sql.extensions:

spark-submit \
  --class com.example.MyRealJob \
  --master yarn \
  --jars invaract-spark-adapter-0.11.0.jar \
  --conf spark.sql.extensions=com.invaract.sparkadapter.InvaractSparkSessionExtension \
  --conf spark.invaract.contract=/path/to/contract.yaml \
  my-real-job.jar

MyRealJob’s source never imports anything from com.invaract.contract/ com.invaract.sparkadapter — a platform team, an orchestration framework, or a CI pipeline can attach contract enforcement to a job whose source they don’t own at all.

Every other spark.invaract.* setting works with this too — the location map, the stricter checks, a notification sink. Configuration Settings lists them all.

Dry-run mode, without a contract at all
Section titled “Dry-run mode, without a contract at all”

Set spark.invaract.dryRun=true instead of spark.invaract.contract to run dry-run mode — no contract needed, nothing ever blocked. With no job code to hand an inferred contract to, it’s logged at WARN instead:
Terminal window
```
spark-submit \
  --conf spark.sql.extensions=com.invaract.sparkadapter.InvaractSparkSessionExtension \
  --conf spark.invaract.dryRun=true \
  ...
```
Publishing notification events too
Section titled “Publishing notification events too”

Add spark.invaract.notifyConfig=<path> — see Configure a Notification Sink for the full mechanism.

Need more control?

Call ContractEnforcementRule.forContract yourself when you need something the conf-driven path can’t express — a contract loaded from somewhere spark.invaract.contract can’t name, a custom LocationResolver, or any other programmatic control. See Install and Configure Invaract Explicitly in Code.

The demo job ./dev/test runs installs its contract the same way, through spark-submit --conf (“Your First Contract” walks through it). It uses code only to put a dry-run draft and a job summary into its own report, which no setting can do.

Handling a rejection

ContractEnforcementRule.forContract throws ContractViolationException from inside Spark’s query analysis, so it surfaces as an ordinary Scala exception around whatever triggered the write (.write.save(...), .write.parquet(...), and so on):

import com.invaract.verification.ContractViolationException
import scala.util.{Failure, Success, Try}

Try {
  outputDf.write.mode("overwrite").parquet(outputPath)
} match {
  case Success(_) => // write succeeded and was verified
  case Failure(e: ContractViolationException) =>
    // e.getMessage is the full four-part explanation: what the contract
    // expects, what the plan contains, why it violates the contract, and
    // how to correct it.
    logger.error(e.getMessage)
    sys.exit(1)
  case Failure(other) => throw other
}

Reporting on a write, not just gating it

ContractEnforcementRule only decides pass/fail — it doesn’t hand you a structured summary of the write it just verified. If you also want the translated plan and its column-level lineage (for a report, a dashboard, or logging), register SparkAdapterListener alongside it — it observes the same writes after they succeed:

import com.invaract.sparkadapter.SparkAdapterListener

val irListener = new SparkAdapterListener
spark.listenerManager.register(irListener)

// ... your write happens here, gated by ContractEnforcementRule ...

// QueryExecutionListener callbacks run asynchronously on Spark's own
// listener thread, so poll briefly if you need the result immediately
// after write() returns.
val result = irListener.lastWrite // Option[TranslationResult]

When the enforcement rule is installed on the same session, lastWrite describes the same plan the rule checked — including seeing through a .checkpoint() upstream of the write, so it shows your real reads rather than an opaque checkpoint node. With no rule installed, it describes the plan exactly as Spark analyzed it.

See Verification vs. Enforcement for why these are two separate mechanisms rather than one.

Version compatibility warnings

Every time you build a SparkSession with ContractEnforcementRule installed (forContract or dry-run mode), it also checks the running Spark version — and, if either is on the classpath, the running Delta Lake/Iceberg version — against this project’s own verified-version list. Running an unverified version doesn’t block your job: verification proceeds normally, but a line like

WARN VersionCompatibilityGuard: Invaract is running against Spark 3.5.1, which is not in
this project's verified-version list (verified: 3.5.6, 3.5.7, 3.5.9). This is advisory
only - verification will continue normally - but Spark 3.5.1's compatibility with
Invaract has not been checked by this project's own CI. See
https://mlltx.github.io/Invaract/reference/spark-version-support/ for the current
verified-version list.

is logged so you have a concrete signal (not a raw stack trace, if something does go wrong) that you’re outside what’s been tested. See Troubleshooting for the two known failure modes this warning would predict.

What gets checked

Only a plan that translates to a recognized write is verified — everything else (reads, .count(), intermediate transformations) is a silent no-op. See Connector Support for which write shapes are recognized today, and Fail-Closed by Default for what happens when your job writes via a shape Invaract doesn’t recognize at all.

Every declared output must be written

A contract that declares an output promises the job writes it. Spark checks each write as it runs, so whether every declared output was written is known only when the job ends: when the application stops, Invaract publishes a FAILED validation with one MISSING_OUTPUT per declared output the job never wrote, and logs it as an error, for example the job never writes declared output 'audit' (out/audit). The job’s other writes have already happened by then; nothing is blocked.

To have the job fail as well, so the scheduler or CI step running it stops what depends on it:

spark-submit \
  --conf spark.sql.extensions=com.invaract.sparkadapter.InvaractSparkSessionExtension \
  --conf spark.invaract.contract=/path/to/contract.yaml \
  --conf spark.invaract.failOnMissingOutput=true \
  ...

The driver then exits with status 1 after Spark has shut down. Two things to know: in YARN cluster mode a failed attempt may be retried (spark.yarn.maxAppAttempts), which runs the job again; and a write Invaract cannot see (an RDD saveAsTextFile) counts as not written. An engine that sees the whole job before it runs, such as Beam, refuses such a job before it starts instead (see Use Invaract with Apache Beam).