Great Expectations data quality gate for S3 files with Kestra

id: gx-data-quality-gate
namespace: company.team

inputs:
  - id: sample_url
    type: URI
    displayName: Sample CSV (manual runs)
    description: CSV validated when the flow is run manually instead of by the S3 trigger.
    defaults: https://huggingface.co/datasets/kestra/datasets/raw/main/csv/orders.csv

  - id: inject_bad_rows
    type: BOOL
    displayName: Inject bad rows (manual runs)
    description: Validate a second copy of the sample with a duplicate id, an
      invalid email, a negative price and a zero quantity, so you can watch the
      gate reject it.
    defaults: true

  - id: load_to_postgres
    type: BOOL
    displayName: Load passing files on manual runs
    description: S3-triggered runs always load passing files. Enable this to also
      load them on manual runs (requires the Postgres secrets).
    defaults: false

  - id: notify_slack
    type: BOOL
    displayName: Send Slack alerts on manual runs
    description: S3-triggered runs always alert on failures. Enable this to also
      alert on manual runs (requires SLACK_WEBHOOK_URL).
    defaults: false

variables:
  bucket: my-data-landing
  region: eu-west-1
  landing_prefix: landing/orders/
  processed_prefix: processed/orders/
  quarantine_prefix: quarantine/orders/
  target_table: public.orders

tasks:
  - id: download_sample
    type: io.kestra.plugin.core.http.Download
    description: On manual runs, fetch the sample CSV that stands in for files
      arriving in S3.
    runIf: "{{ trigger is not defined }}"
    uri: "{{ inputs.sample_url }}"

  - id: plan_batch
    type: io.kestra.plugin.core.output.OutputValues
    description: Build the list of files to validate, either the new S3 objects or
      the demo files, and record whether the run came from S3.
    values:
      triggered: "{{ trigger is defined }}"
      files: >-
        {{ trigger is defined ? trigger.objects : (inputs.inject_bad_rows ? [
        {'key': vars.landing_prefix ~ 'orders.csv', 'uri':
        outputs.download_sample.uri, 'inject': false}, {'key':
        vars.landing_prefix ~ 'orders_with_bad_rows.csv', 'uri':
        outputs.download_sample.uri, 'inject': true} ] : [ {'key':
        vars.landing_prefix ~ 'orders.csv', 'uri': outputs.download_sample.uri,
        'inject': false} ]) }}

  - id: validate_files
    type: io.kestra.plugin.core.flow.Loop
    description: Validate every file in parallel and route it based on its Great
      Expectations result.
    values: "{{ outputs.plan_batch.values.files }}"
    concurrencyLimit: 4
    tasks:
      - id: validate
        type: io.kestra.plugin.scripts.python.Script
        description: Run the expectation suite against the file with Great Expectations
          and output the verdict and the failing expectations.
        containerImage: python:3.12-slim
        beforeCommands:
          - pip install -q --root-user-action=ignore great_expectations==1.23.2
        env:
          INJECT_BAD_ROWS: "{{ fromJson(item.value).inject ?? false }}"
          FILE_KEY: "{{ fromJson(item.value).key }}"
          GX_ANALYTICS_ENABLED: "false"
        inputFiles:
          data.csv: "{{ fromJson(item.value).uri }}"
          suite.json: |
            {
              "name": "orders",
              "expectations": [
                {"type": "expect_table_columns_to_match_ordered_list", "kwargs": {"column_list": ["order_id", "customer_name", "customer_email", "product_id", "price", "quantity", "total"]}},
                {"type": "expect_column_values_to_not_be_null", "kwargs": {"column": "order_id"}},
                {"type": "expect_column_values_to_be_unique", "kwargs": {"column": "order_id"}},
                {"type": "expect_column_values_to_match_regex", "kwargs": {"column": "customer_email", "regex": "^[^@\\s]+@[^@\\s]+\\.[A-Za-z]{2,}$"}},
                {"type": "expect_column_values_to_not_be_null", "kwargs": {"column": "price"}},
                {"type": "expect_column_values_to_be_between", "kwargs": {"column": "price", "min_value": 0, "strict_min": true}},
                {"type": "expect_column_values_to_not_be_null", "kwargs": {"column": "quantity"}},
                {"type": "expect_column_values_to_be_between", "kwargs": {"column": "quantity", "min_value": 1, "max_value": 100}}
              ]
            }
        script: |
          import json
          import os

          import great_expectations as gx
          import pandas as pd
          from great_expectations.expectations.expectation_configuration import ExpectationConfiguration

          try:
              df = pd.read_csv("data.csv")
          except Exception as error:
              outputs = {"passed": False, "rows": 0, "evaluated": 0, "failed_count": 0, "failed": "", "summary": f"unreadable CSV: {error}"}
              print(f"{os.environ['FILE_KEY']}: FAILED {outputs}")
              print("::" + json.dumps({"outputs": outputs}) + "::")
              raise SystemExit(0)

          if os.environ["INJECT_BAD_ROWS"].lower() == "true":
              bad_rows = pd.DataFrame([
                  {"order_id": 1, "customer_name": "Duplicate Id", "customer_email": "dup@example.com", "product_id": 1, "price": 10.0, "quantity": 1, "total": 10.0},
                  {"order_id": 9001, "customer_name": "Bad Email", "customer_email": "not-an-email", "product_id": 2, "price": 10.0, "quantity": 1, "total": 10.0},
                  {"order_id": 9002, "customer_name": "Negative Price", "customer_email": "neg@example.com", "product_id": 3, "price": -5.0, "quantity": 1, "total": -5.0},
                  {"order_id": 9003, "customer_name": "Bulk Typo", "customer_email": "bulk@example.com", "product_id": 4, "price": 10.0, "quantity": 500, "total": 5000.0},
              ])
              df = pd.concat([df, bad_rows], ignore_index=True)

          with open("suite.json") as f:
              spec = json.load(f)
          suite = gx.ExpectationSuite(name=spec["name"])
          for expectation in spec["expectations"]:
              suite.add_expectation_configuration(
                  ExpectationConfiguration(type=expectation["type"], kwargs=expectation["kwargs"])
              )

          context = gx.get_context(mode="ephemeral")
          batch_definition = (
              context.data_sources.add_pandas("files")
              .add_dataframe_asset("file")
              .add_batch_definition_whole_dataframe("whole_file")
          )
          result = batch_definition.get_batch(batch_parameters={"dataframe": df}).validate(suite)

          failed = []
          for r in result.results:
              if not r.success:
                  column = r.expectation_config.kwargs.get("column", "table")
                  unexpected = r.result.get("unexpected_count")
                  detail = f" ({unexpected} rows)" if unexpected is not None else ""
                  failed.append(f"{r.expectation_config.type} on {column}{detail}")

          outputs = {
              "passed": result.success,
              "rows": len(df),
              "evaluated": result.statistics["evaluated_expectations"],
              "failed_count": len(failed),
              "failed": "; ".join(failed),
              "summary": f"{len(failed)} of {result.statistics['evaluated_expectations']} expectations failed: " + "; ".join(failed),
          }
          print(f"{os.environ['FILE_KEY']}: {'PASSED' if result.success else 'FAILED'} {outputs}")
          print("::" + json.dumps({"outputs": outputs}) + "::")

      - id: route_file
        type: io.kestra.plugin.core.flow.If
        description: Load and archive files that pass; quarantine and report files that
          fail.
        condition: "{{ outputs.validate.vars.passed }}"
        then:
          - id: load_to_postgres
            type: io.kestra.plugin.jdbc.postgresql.CopyIn
            description: Append the validated rows to the target table.
            runIf: "{{ (outputs.plan_batch.values.triggered | string) == 'true' or
              inputs.load_to_postgres }}"
            url: "{{ secret('POSTGRES_URL') }}"
            username: "{{ secret('POSTGRES_USER') }}"
            password: "{{ secret('POSTGRES_PASSWORD') }}"
            format: CSV
            header: true
            table: "{{ vars.target_table }}"
            from: "{{ fromJson(item.value).uri }}"

          - id: archive_file
            type: io.kestra.plugin.aws.s3.Upload
            description: Keep a copy of the loaded file under the processed prefix.
            runIf: "{{ (outputs.plan_batch.values.triggered | string) == 'true' }}"
            accessKeyId: "{{ secret('AWS_ACCESS_KEY_ID') }}"
            secretKeyId: "{{ secret('AWS_SECRET_ACCESS_KEY') }}"
            region: "{{ vars.region }}"
            bucket: "{{ vars.bucket }}"
            key: "{{ vars.processed_prefix ~ (execution.startDate | date('yyyy-MM-dd')) ~
              '/' ~ execution.id ~ '_' ~ (fromJson(item.value).key | split('/')
              | last) }}"
            from: "{{ fromJson(item.value).uri }}"
        else:
          - id: log_failure
            type: io.kestra.plugin.core.log.Log
            description: Record why the file was rejected, even when Slack is off.
            level: WARN
            message: "Rejected {{ fromJson(item.value).key }}: {{
              outputs.validate.vars.summary }}"

          - id: quarantine_file
            type: io.kestra.plugin.aws.s3.Upload
            description: Copy the rejected file to the quarantine prefix for inspection.
            runIf: "{{ (outputs.plan_batch.values.triggered | string) == 'true' }}"
            accessKeyId: "{{ secret('AWS_ACCESS_KEY_ID') }}"
            secretKeyId: "{{ secret('AWS_SECRET_ACCESS_KEY') }}"
            region: "{{ vars.region }}"
            bucket: "{{ vars.bucket }}"
            key: "{{ vars.quarantine_prefix ~ (execution.startDate | date('yyyy-MM-dd')) ~
              '/' ~ execution.id ~ '_' ~ (fromJson(item.value).key | split('/')
              | last) }}"
            from: "{{ fromJson(item.value).uri }}"

          - id: alert_slack
            type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
            description: Tell the data team which expectations the file failed.
            runIf: "{{ (outputs.plan_batch.values.triggered | string) == 'true' or
              inputs.notify_slack }}"
            url: "{{ secret('SLACK_WEBHOOK_URL') }}"
            payload: |
              {
                "text": {{ ('Data quality gate rejected `' ~ fromJson(item.value).key ~ '`: ' ~ outputs.validate.vars.summary ~ '. Execution: ' ~ execution.id) | toJson }}
              }

errors:
  - id: alert_failed_run
    type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
    description: Alert when the run itself fails, because the trigger has already
      removed the files from the landing prefix.
    runIf: "{{ trigger is defined or inputs.notify_slack }}"
    url: "{{ secret('SLACK_WEBHOOK_URL') }}"
    payload: |
      {
        "text": {{ ('Data quality gate execution ' ~ execution.id ~ ' failed before routing all files. They are kept in Kestra internal storage: fix the cause and restart the execution.') | toJson }}
      }

triggers:
  - id: new_files
    type: io.kestra.plugin.aws.s3.Trigger
    description: Start the gate when new CSV files land under the landing prefix.
      Enable after setting the AWS secrets.
    disabled: true
    interval: PT1M
    accessKeyId: "{{ secret('AWS_ACCESS_KEY_ID') }}"
    secretKeyId: "{{ secret('AWS_SECRET_ACCESS_KEY') }}"
    region: "{{ vars.region }}"
    bucket: "{{ vars.bucket }}"
    prefix: "{{ vars.landing_prefix }}"
    regexp: ".*\\.csv$"
    filter: FILES
    maxKeys: 20
    action: DELETE

Bad files should never reach the warehouse. This flow puts a Great Expectations (GX) gate between an S3 landing prefix and Postgres. Every new file is checked against an expectation suite, in parallel:

Files that pass are appended to the target table and archived under a processed prefix.
Files that fail are moved to a quarantine prefix, are never loaded, and Slack gets the list of failing expectations with the number of offending rows.

The expectation suite is plain JSON inside the flow, so you can change the rules without touching the Python code.

It runs with no setup. On a manual run the flow validates a public CSV of 100 orders and a copy of it with four bad rows, so you can watch one file pass and the other get rejected. S3 copies only happen on S3-triggered runs; loading and Slack also run on manual runs when you switch them on with inputs.

This blueprint was created by manojgosavi.

How it works

new_files (io.kestra.plugin.aws.s3.Trigger) polls the landing prefix every minute for .csv files and starts an execution with up to 20 of them. The files are downloaded to Kestra's internal storage and deleted from the landing prefix (action: DELETE), so a file is never picked up twice.
On manual runs, download_sample (io.kestra.plugin.core.http.Download) fetches the sample CSV instead.
plan_batch (io.kestra.plugin.core.output.OutputValues) builds the list of files (the S3 objects, or the two demo files) and records whether the run came from S3.
validate_files (io.kestra.plugin.core.flow.Loop) processes up to 4 files at a time. For each file:
- validate (io.kestra.plugin.scripts.python.Script, great_expectations==1.23.2) runs the suite with an ephemeral GX context and outputs passed, rows, evaluated, failed_count, failed and a one-line summary. A file that can't be read as CSV fails with the summary unreadable CSV: <error>.
- route_file (io.kestra.plugin.core.flow.If):
  - Passed: load_to_postgres (io.kestra.plugin.jdbc.postgresql.CopyIn) appends the rows to public.orders and archive_file (io.kestra.plugin.aws.s3.Upload) writes the file to processed/orders/<date>/<execution id>_<file name>.
  - Failed: log_failure (io.kestra.plugin.core.log.Log) logs the reasons, quarantine_file (io.kestra.plugin.aws.s3.Upload) writes the file to quarantine/orders/<date>/<execution id>_<file name> and alert_slack (io.kestra.plugin.slack.notifications.SlackIncomingWebhook) posts the failing expectations.
If the execution itself fails, alert_failed_run (io.kestra.plugin.slack.notifications.SlackIncomingWebhook) posts to Slack so the files are not silently stuck in internal storage.

The expectation suite

The suite.json input file of validate checks that:

the columns are exactly order_id, customer_name, customer_email, product_id, price, quantity, total, in this order (CopyIn loads columns by position),
order_id is never null and is unique,
customer_email looks like an email address,
price is never null and is greater than 0,
quantity is never null and is between 1 and 100.

Each entry is an expectation type and its kwargs, as listed in the Expectation Gallery. Add, remove or tune entries to match your data.

Inputs

sample_url (URI, default the public orders CSV): the file validated on manual runs.
inject_bad_rows (BOOL, default true): also validate orders_with_bad_rows.csv, a copy of the sample with a duplicate id, an invalid email, a negative price and a quantity of 500.
load_to_postgres (BOOL, default false): load passing files on manual runs too.
notify_slack (BOOL, default false): send Slack alerts on manual runs too.

Prerequisites

A Kestra worker that can run Docker containers. The script uses python:3.12-slim and installs the pinned GX version for each file, which takes about 20 seconds. For high volumes, build an image with GX preinstalled and drop beforeCommands.

For S3-triggered runs: an S3 bucket, and a Postgres table whose columns match the CSV header, for example:

CREATE TABLE public.orders (order_id int, customer_name text, customer_email text, product_id int, price numeric, quantity int, total numeric);

Secrets

Only needed for S3-triggered runs, or when you enable loading or alerts on manual runs:

AWS_ACCESS_KEY_ID and AWS_SECRET_ACCESS_KEY: read and delete objects under the landing prefix, write under the processed and quarantine prefixes.
POSTGRES_URL (for example jdbc:postgresql://host:5432/warehouse), POSTGRES_USER, POSTGRES_PASSWORD: insert into the target table.
SLACK_WEBHOOK_URL: Slack incoming webhook for quarantine alerts.

Quick start

Run the flow. In the Logs, orders.csv passes all 8 expectations and orders_with_bad_rows.csv fails 4 of them and is rejected with a WARN log line. On S3-triggered runs it would also be copied to quarantine and reported to Slack.
Run it again with inject_bad_rows: false. Only orders.csv is validated, and it passes.
For production, create the table, set the secrets, change bucket, region, the prefixes and target_table in variables, and set disabled: false on the new_files trigger. Then drop a CSV under landing/orders/.

Expected outputs

Per file: outputs.validate.vars.passed, rows, evaluated, failed_count, failed (for example expect_column_values_to_be_unique on order_id (2 rows)) and summary.
Passing files: rows in public.orders and a copy under processed/orders/<date>/.
Failing files: a WARN log line and, on S3-triggered runs, a copy under quarantine/orders/<date>/ and a Slack message.

Things to know

A file is either loaded in full or not at all. The gate does not split good rows from bad ones.
The trigger deletes files from the landing prefix as soon as they are downloaded, before validation. If an execution fails (for example a network error while installing GX, or a Postgres outage), the files only exist in Kestra's internal storage: alert_failed_run tells you, and restarting the execution from the Executions page processes them again.
Archived and quarantined copies are prefixed with the date and execution id, so files with the same name never overwrite each other.
To re-process a quarantined file after fixing it, copy it back under the landing prefix.

How to extend

Keep one suite per dataset in namespace files and pick it from the file's prefix.
Build GX Data Docs from the validation result and publish them as namespace files.
Replace CopyIn with a load into Snowflake, BigQuery or Databricks.

Quarantine bad files from S3 with a Great Expectations data quality gate

How it works

The expectation suite

Inputs

Prerequisites

Secrets

Quick start

Expected outputs

Things to know

How to extend

Links

Tasks

Orchestrate with Kestra

Share this Blueprint

Create a Blueprint

New to Kestra?

Quarantine bad files from S3 with a Great Expectations data quality gate

How it works

The expectation suite

Inputs

Prerequisites

Secrets

Quick start

Expected outputs

Things to know

How to extend

Links

Tasks

Orchestrate with Kestra

Share this Blueprint

Create a Blueprint

More Data Blueprints

Recheck a Sifflet Rule After an Incident Fix

Archive Slack Channel History into BigQuery

Schedule a Daily Sifflet Observability Sweep

Gate a Dashboard Publish on a Sifflet Rule

Catch Link Rot Before Your Users and Your Crawler Do

Validate Releases with a Sifflet Rule Run

New to Kestra?