Automated OpenLineage Metadata Extraction and Marquez Governance Pipeline

id: openlineage-marquez-data-governance
namespace: company.team

description: |
  Automated data governance and OpenLineage metadata extraction pipeline with in-process
  DuckDB analytical SQL transformations, Marquez lineage ingestion, and Slack alerting.

inputs:
  - id: pipeline_id
    type: STRING
    defaults: customer-revenue-mart-v1
    description: Unique identifier of the analytical data pipeline.

  - id: marquez_endpoint
    type: STRING
    defaults: "http://marquez:5000/api/v1/lineage"
    description: HTTP URL of the Marquez or OpenLineage ingestion API.

  - id: duckdb_database_path
    type: STRING
    defaults: "/tmp/kestra_analytics.duckdb"
    description: Filesystem path for local DuckDB storage.

  - id: alert_on_schema_drift
    type: BOOLEAN
    defaults: true
    description: Whether to pause the flow and trigger Slack alerts if schema drift
      is detected.

triggers:
  - id: schedule_trigger
    type: io.kestra.plugin.core.trigger.Schedule
    cron: "0 4 * * *"

  - id: webhook_trigger
    type: io.kestra.plugin.core.trigger.Webhook
    key: "{{ secret('LINEAGE_WEBHOOK_KEY') }}"

tasks:
  - id: setup_governance_context
    type: io.kestra.plugin.core.log.Log
    message: "Initializing OpenLineage metadata extraction for pipeline {{
      inputs.pipeline_id }} against endpoint {{ inputs.marquez_endpoint }}"

  - id: run_duckdb_transformation
    type: io.kestra.plugin.jdbc.duckdb.Queries
    url: "jdbc:duckdb:{{ inputs.duckdb_database_path }}"
    sql: |
      CREATE TABLE IF NOT EXISTS raw_customer_transactions (
          transaction_id VARCHAR,
          customer_id VARCHAR,
          amount DOUBLE,
          currency VARCHAR,
          created_at TIMESTAMP
      );

      INSERT INTO raw_customer_transactions VALUES
          ('tx-1001', 'cust-01', 149.50, 'USD', CURRENT_TIMESTAMP),
          ('tx-1002', 'cust-02', 890.00, 'USD', CURRENT_TIMESTAMP),
          ('tx-1003', 'cust-01', 210.25, 'USD', CURRENT_TIMESTAMP);

      CREATE TABLE IF NOT EXISTS curated_customer_revenue_mart AS
      SELECT
          customer_id,
          COUNT(transaction_id) AS total_transactions,
          SUM(amount) AS total_revenue_usd,
          CURRENT_DATE AS mart_date
      FROM raw_customer_transactions
      GROUP BY customer_id;

  - id: extract_openlineage_facets
    type: io.kestra.plugin.scripts.python.Commands
    containerImage: python:3.11-slim
    beforeCommands:
      - pip install --no-cache-dir pydantic
    commands:
      - python extract_lineage.py
    inputFiles:
      extract_lineage.py: |
        import datetime
        import json
        import os
        import uuid

        run_id = str(uuid.uuid4())
        job_name = "{{ inputs.pipeline_id }}"
        job_namespace = "company.team"
        event_time = datetime.datetime.now(datetime.timezone.utc).isoformat()

        openlineage_event = {
            "eventType": "COMPLETE",
            "eventTime": event_time,
            "run": {
                "runId": run_id,
                "facets": {
                    "nominalTime": {
                        "_producer": "https://github.com/kestra-io/kestra",
                        "_schemaURL": "https://openlineage.io/spec/facets/1-0-0/NominalTimeRunFacet.json",
                        "nominalStartTime": event_time
                    }
                }
            },
            "job": {
                "namespace": job_namespace,
                "name": job_name,
                "facets": {}
            },
            "inputs": [
                {
                    "namespace": "duckdb://{{ inputs.duckdb_database_path }}",
                    "name": "raw_customer_transactions",
                    "facets": {
                        "schema": {
                            "_producer": "https://github.com/kestra-io/kestra",
                            "_schemaURL": "https://openlineage.io/spec/facets/1-0-0/SchemaDatasetFacet.json",
                            "fields": [
                                {"name": "transaction_id", "type": "VARCHAR"},
                                {"name": "customer_id", "type": "VARCHAR"},
                                {"name": "amount", "type": "DOUBLE"},
                                {"name": "currency", "type": "VARCHAR"},
                                {"name": "created_at", "type": "TIMESTAMP"}
                            ]
                        }
                    }
                }
            ],
            "outputs": [
                {
                    "namespace": "duckdb://{{ inputs.duckdb_database_path }}",
                    "name": "curated_customer_revenue_mart",
                    "facets": {
                        "schema": {
                            "_producer": "https://github.com/kestra-io/kestra",
                            "_schemaURL": "https://openlineage.io/spec/facets/1-0-0/SchemaDatasetFacet.json",
                            "fields": [
                                {"name": "customer_id", "type": "VARCHAR"},
                                {"name": "total_transactions", "type": "BIGINT"},
                                {"name": "total_revenue_usd", "type": "DOUBLE"},
                                {"name": "mart_date", "type": "DATE"}
                            ]
                        }
                    }
                }
            ],
            "producer": "https://github.com/kestra-io/kestra"
        }

        with open("openlineage_payload.json", "w") as f:
            json.dump(openlineage_event, f, indent=2)

        outputs = {
            "run_id": run_id,
            "job_name": job_name,
            "input_dataset_count": 1,
            "output_dataset_count": 1,
            "passed_lineage_validation": True,
            "payload_size_bytes": os.path.getsize("openlineage_payload.json")
        }
        print("::" + json.dumps({"outputs": outputs}) + "::")
    outputFiles:
      - "openlineage_payload.json"

  - id: emit_openlineage_to_marquez
    type: io.kestra.plugin.core.http.Request
    uri: "{{ inputs.marquez_endpoint }}"
    method: POST
    contentType: application/json
    body: "{{
      read(outputs.extract_openlineage_facets.outputFiles['openlineage_payload.\
      json']) }}"
    headers:
      Authorization: "Bearer {{ secret('MARQUEZ_API_KEY') }}"
      User-Agent: "Kestra-OpenLineage-Governance-Agent/1.0"
    timeout: PT30S

  - id: verify_marquez_ingestion
    type: io.kestra.plugin.core.log.Log
    message: "Lineage event for run {{
      outputs.extract_openlineage_facets.vars.run_id }} successfully emitted to
      Marquez. HTTP Status: {{ outputs.emit_openlineage_to_marquez.status }}"

  - id: check_governance_guardrail
    type: io.kestra.plugin.core.flow.If
    condition: "{{ outputs.extract_openlineage_facets.vars.passed_lineage_validation
      == true }}"
    then:
      - id: log_governance_passed
        type: io.kestra.plugin.core.log.Log
        message: "Data governance validation passed. Output datasets and schema facets
          registered."

      - id: notify_slack_success
        type: io.kestra.plugin.notifications.slack.SlackIncomingWebhook
        url: "{{ secret('SLACK_WEBHOOK_URL') }}"
        payload: |
          {
            "text": "Data Governance Pipeline Passed: Job {{ inputs.pipeline_id }} lineage facets successfully recorded in Marquez."
          }

    else:
      - id: log_governance_failure
        type: io.kestra.plugin.core.log.Log
        message: "Data governance check failed. Schema drift or lineage validation
          exception detected."

      - id: notify_slack_alert
        type: io.kestra.plugin.notifications.slack.SlackIncomingWebhook
        url: "{{ secret('SLACK_WEBHOOK_URL') }}"
        payload: |
          {
            "text": "Data Governance Alert: Job {{ inputs.pipeline_id }} failed lineage validation. Manual compliance sign-off required."
          }

      - id: require_compliance_officer_approval
        type: io.kestra.plugin.core.flow.Pause
        tasks:
          - id: log_pause_status
            type: io.kestra.plugin.core.log.Log
            message: "Pipeline paused. Awaiting Data Governance officer review and
              approval."