JSON Schema Contract Validation & Parquet Quarantine

id: json-schema-data-contract-validation-and-parquet-quarantine
namespace: company.data

inputs:
  - id: raw_json_payload
    type: STRING
    displayName: Raw JSON Event Batch
    description: Incoming JSON array of telemetry or transaction event records to
      validate and ingest.
    defaults: |
      [
        {
          "event_id": "a5c7f8e2-3b4d-4e5f-9a1b-2c3d4e5f6a7b",
          "timestamp": "2026-10-17T12:00:00Z",
          "tenant_id": "tenant-acme-corp",
          "event_type": "order_placed",
          "payload": {
            "order_id": "ORD-98231",
            "amount": 249.99,
            "currency": "USD",
            "customer_id": "CUST-4412"
          }
        },
        {
          "event_id": "b6d8a9f3-4c5e-5f6a-0b2c-3d4e5f6a7b8c",
          "timestamp": "2026-10-17T12:01:15Z",
          "tenant_id": "tenant-globex-ind",
          "event_type": "payment_processed",
          "payload": {
            "order_id": "ORD-98232",
            "amount": 119.50,
            "currency": "EUR",
            "customer_id": "CUST-8821"
          }
        }
      ]

  - id: json_schema_contract
    type: STRING
    displayName: JSON Schema Contract Definition
    description: Strict JSON Schema (Draft-07) defining mandatory fields, formats,
      regex patterns, and type assertions.
    defaults: |
      {
        "$schema": "http://json-schema.org/draft-07/schema#",
        "title": "TransactionalEventBatch",
        "type": "array",
        "items": {
          "type": "object",
          "required": [
            "event_id",
            "timestamp",
            "tenant_id",
            "event_type",
            "payload"
          ],
          "properties": {
            "event_id": {
              "type": "string",
              "format": "uuid"
            },
            "timestamp": {
              "type": "string",
              "format": "date-time"
            },
            "tenant_id": {
              "type": "string",
              "minLength": 3
            },
            "event_type": {
              "type": "string",
              "enum": [
                "order_placed",
                "order_cancelled",
                "payment_processed",
                "inventory_reserved"
              ]
            },
            "payload": {
              "type": "object",
              "required": [
                "order_id",
                "amount",
                "currency",
                "customer_id"
              ],
              "properties": {
                "order_id": {
                  "type": "string"
                },
                "amount": {
                  "type": "number",
                  "minimum": 0.01
                },
                "currency": {
                  "type": "string",
                  "pattern": "^[A-Z]{3}$"
                },
                "customer_id": {
                  "type": "string"
                }
              },
              "additionalProperties": false
            }
          },
          "additionalProperties": false
        }
      }

  - id: schema_version
    type: STRING
    displayName: Schema Contract Version
    description: Semantic version of the data contract governing this ingestion stream.
    defaults: "1.2.0"

  - id: fail_on_contract_breach
    type: BOOL
    displayName: Fail Pipeline on Contract Breach
    description: When true, marks the workflow execution as FAILED after isolating
      records into quarantine. When false, marks execution successful while
      preserving quarantine artifacts.
    defaults: false

  - id: send_slack_alert
    type: BOOL
    displayName: Send Slack Alert on Breach
    description: Dispatches an incident report to Slack when malformed records or
      contract violations occur.
    defaults: true

  - id: slack_webhook_url
    type: STRING
    displayName: Slack Webhook URL
    description: Webhook URL used to notify data engineering on contract drift.
    defaults: "{{ secret('SLACK_WEBHOOK_URL') }}"

triggers:
  - id: webhook_event_intake
    type: io.kestra.plugin.core.trigger.Webhook
    key: data-contract-intake
    description: Real-time event ingestion webhook receiving JSON payloads from
      microservices or message queues.

  - id: nightly_batch_schedule
    type: io.kestra.plugin.core.trigger.Schedule
    cron: "0 2 * * *"
    description: Nightly scheduled execution for automated contract audit and
      synthetic pipeline verification.

tasks:
  - id: stage_incoming_payload
    type: io.kestra.plugin.core.storage.Write
    description: Persists incoming batch data into Kestra internal storage for
      isolated validation and transformation.
    extension: .json
    content: "{{ trigger.body is defined ? trigger.body : inputs.raw_json_payload }}"

  - id: validate_contract
    type: io.kestra.plugin.serdes.json.ValidateJsonSchema
    description: Evaluates batch against the JSON Schema contract, capturing
      granular field constraint violations without throwing execution
      exceptions.
    from: "{{ outputs.stage_incoming_payload.uri }}"
    schema: "{{ inputs.json_schema_contract }}"
    defaultDialect: DRAFT_7
    failOnInvalid: false

  - id: contract_conformance_switch
    type: io.kestra.plugin.core.flow.Switch
    description: Bifurcates workflow routing based on whether the incoming records
      strictly matched the data contract.
    value: "{{ outputs.validate_contract.valid ? 'COMPLIANT' : 'BREACHED' }}"
    cases:
      COMPLIANT:
        - id: convert_json_to_ion
          type: io.kestra.plugin.serdes.json.JsonToIon
          description: Converts compliant JSON array records into high-performance Amazon
            Ion binary format.
          from: "{{ outputs.stage_incoming_payload.uri }}"

        - id: compile_parquet
          type: io.kestra.plugin.serdes.parquet.IonToParquet
          description: Serializes Ion stream into compressed columnar Apache Parquet
            format ready for data lake landing.
          from: "{{ outputs.convert_json_to_ion.uri }}"

        - id: log_landing_metrics
          type: io.kestra.plugin.core.log.Log
          description: Records ingestion telemetry, record count, and artifact references.
          message: |
            DATA CONTRACT COMPLIANT: Ingested batch passed validation against contract v{{ inputs.schema_version }}.
            Successfully transformed {{ outputs.convert_json_to_ion.size }} records into Parquet format:
            Parquet URI: {{ outputs.compile_parquet.uri }}
            Total Records Compiled: {{ outputs.compile_parquet.size }}

      BREACHED:
        - id: quarantine_payload
          type: io.kestra.plugin.core.storage.Write
          description: Isolates malformed records and schema error diagnostics into a
            dead-letter quarantine artifact.
          extension: .json
          content: |
            {
              "quarantine_timestamp": "{{ execution.startDate }}",
              "execution_id": "{{ execution.id }}",
              "contract_version": "{{ inputs.schema_version }}",
              "violation_count": {{ outputs.validate_contract.errors | length }},
              "violations": {{ toJson(outputs.validate_contract.errors) }},
              "raw_payload_uri": "{{ outputs.stage_incoming_payload.uri }}"
            }

        - id: notify_contract_breach
          type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
          description: Alerts data platform teams on contract violations with exact schema
            paths and offending fields.
          url: "{{ inputs.slack_webhook_url }}"
          runIf: "{{ inputs.send_slack_alert }}"
          payload: |
            {
              "text": "*DATA CONTRACT BREACH DETECTED* in flow `{{ flow.id }}` (Execution `{{ execution.id }}`)",
              "blocks": [
                {
                  "type": "header",
                  "text": {
                    "type": "plain_text",
                    "text": "Data Contract Validation Failure",
                    "emoji": true
                  }
                },
                {
                  "type": "section",
                  "fields": [
                    {
                      "type": "mrkdwn",
                      "text": "*Contract Version:* `{{ inputs.schema_version }}`"
                    },
                    {
                      "type": "mrkdwn",
                      "text": "*Violations Detected:* `{{ outputs.validate_contract.errors | length }}`"
                    },
                    {
                      "type": "mrkdwn",
                      "text": "*Execution ID:* `{{ execution.id }}`"
                    },
                    {
                      "type": "mrkdwn",
                      "text": "*Quarantine Artifact:* `{{ outputs.quarantine_payload.uri }}`"
                    }
                  ]
                },
                {
                  "type": "section",
                  "text": {
                    "type": "mrkdwn",
                    "text": "*Schema Error Summary:* Total violations found: {{ outputs.validate_contract.errors | length }}. Check the quarantine artifact for individual failing JSON paths and error details."
                  }
                }
              ]
            }

        - id: enforce_contract_failure
          type: io.kestra.plugin.core.flow.If
          description: Optional gate to fail pipeline execution when strict upstream SLA
            guarantees are enforced.
          condition: "{{ inputs.fail_on_contract_breach }}"
          then:
            - id: abort_execution
              type: io.kestra.plugin.core.execution.Fail
              description: Halts workflow with non-zero exit code due to contract violation.
              errorMessage: "Contract breach: {{ outputs.validate_contract.errors | length }}
                schema violations detected in landing batch. Malformed payload
                quarantined at {{ outputs.quarantine_payload.uri }}."

Data pipelines frequently suffer from silent schema drift, missing mandatory fields, and unexpected data types introduced by upstream microservices or third-party webhooks. When unvalidated JSON lands directly into data lakes or analytical warehouses, downstream dbt transformations, Spark jobs, and Trino/DuckDB queries fail abruptly hours later during production reporting cycles.

This workflow implements a declarative Data Contract Gate for streaming and batch event ingestion. It validates incoming JSON payloads against a strict JSON Schema (Draft-07), bifurcates compliant data from malformed records, serializes verified events into columnar Apache Parquet format, and routes contract breaches into an isolated quarantine storage location accompanied by diagnostic Slack alerts.

How it works

Batch Ingestion & Staging: An incoming JSON payload is received either via real-time Webhook intake (webhook_event_intake) or automated nightly schedule (nightly_batch_schedule). The stage_incoming_payload task writes the raw batch into Kestra internal storage.
Contract Evaluation: The validate_contract task uses io.kestra.plugin.serdes.json.ValidateJsonSchema to check the dataset against the declared JSON Schema contract without failing execution (failOnInvalid: false). It enforces strict field requirements, UUID formats, ISO-8601 timestamps, enum values, and numeric thresholds.
Operational Conformance Routing: A Switch task evaluates outputs.validate_contract.valid:
- COMPLIANT Branch:
  - Converts verified JSON objects into Amazon Ion binary representation via io.kestra.plugin.serdes.json.JsonToIon.
  - Compiles the Ion stream into Snappy-compressed columnar Apache Parquet using io.kestra.plugin.serdes.parquet.IonToParquet.
  - Logs landing metrics including total records processed, Parquet storage URI, and contract version.
- BREACHED Branch:
  - Quarantines the invalid payload into internal storage with complete violation metadata (failing JSON paths, error codes, and execution ID) via quarantine_payload.
  - Dispatches an actionable alert to Slack containing exact schema failure locations (instanceLocation) and violated rules.
  - Optionally fails the pipeline via enforce_contract_failure if strict enforcement (fail_on_contract_breach: true) is required.

What you get

Zero Silent Lakehouse Corruption: Guarantees that only structurally validated records conforming to data contracts reach curated Parquet storage layers.
Dead-Letter Quarantine: Prevents unparseable payloads from being discarded; corrupt records are stored with full context for inspection and reprocessing.
Native Columnar Serialization: Transforms raw nested JSON directly into compact, query-optimized Apache Parquet files without requiring external compute clusters.
Field-Level Error Diagnostics: Identifies the exact offending JSON path (e.g. [1].payload.amount) and constraint failure reason rather than generic parsing errors.

Who it's for

Data Platform Engineers: Protecting central S3, MinIO, or Google Cloud Storage data lakes from upstream schema changes.
Analytics Engineers: Ensuring dbt models and warehouse tables ingest only validated, high-quality event streams.
Backend & Event-Driven Architects: Implementing Shift-Left data contracts between microservices and analytics teams.

Why orchestrate this with Kestra

Building data contract gates with custom cron scripts or Lambda functions requires writing fragile validation code, managing temporary disk buffers for format conversion, and maintaining separate alerting scripts. Kestra orchestrates the entire contract lifecycle declaratively:

Native SerDes Plugins: High-performance streaming conversions (ValidateJsonSchema, JsonToIon, IonToParquet) execute inside Kestra without requiring Python dependencies or heavyweight Spark clusters.
Dynamic Flow Routing: Seamlessly branches between lakehouse compilation and dead-letter quarantine using built-in condition logic.
Multi-Modal Triggers: Accepts streaming payloads via Webhook triggers while maintaining scheduled backfill capabilities on the same pipeline.

Prerequisites

A running Kestra instance with access to the plugin-serdes and plugin-notifications plugins (included in standard Kestra distributions).
Network access from Kestra to your Slack workspace (for incident notifications).

Secrets

SLACK_WEBHOOK_URL: Slack Incoming Webhook URL used to publish contract breach incident notifications to your data operations channel.

Quick start

Configure the SLACK_WEBHOOK_URL secret in your Kestra namespace or key-value store.
Import this workflow into your Kestra instance.
Run an initial test execution using the default compliant payload to verify Parquet compilation.
Test the quarantine mechanism by submitting an invalid payload (e.g., negative amount or missing event_id) to watch the workflow route records to dead-letter quarantine and dispatch a Slack alert.
Wire upstream microservices or message queues to post JSON batches to the generated Webhook URL.

How to extend

Cloud Object Storage Landing: Add an io.kestra.plugin.aws.s3.Upload or io.kestra.plugin.gcp.gcs.Upload task to ship compiled Parquet files directly to production S3/GCS buckets.
Confluent Schema Registry Integration: Swap the inline schema for registry configuration in ValidateJsonSchema to pull versioned schemas dynamically from Karapace or Confluent Schema Registry.
Automated Replay: Implement an approval-gated subflow that allows data engineers to patch quarantined JSON records and re-inject them into the pipeline.

JSON Schema Data Contract Validation and Parquet Lake Landing Quarantine

How it works

What you get

Who it's for

Why orchestrate this with Kestra

Prerequisites

Secrets

Quick start

How to extend

Links

Tasks

Orchestrate with Kestra

Share this Blueprint

Create a Blueprint

New to Kestra?

JSON Schema Data Contract Validation and Parquet Lake Landing Quarantine

How it works

What you get

Who it's for

Why orchestrate this with Kestra

Prerequisites

Secrets

Quick start

How to extend

Links

Tasks

Orchestrate with Kestra

Share this Blueprint

Create a Blueprint

More Data Blueprints

Hourly OpenSearch Error Spike Gate with Discord Alerting

Use OpenSearch as a Config Store with Put and Get Roundtrip

OpenSearch ISM Policy Compliance Gate

Daily OpenSearch Error Digest to Slack

Search OpenAlex for Related Scholarly Works

Automated OpenLineage Metadata Extraction and Marquez Governance Pipeline

New to Kestra?