Delta Lake table health gate with Kestra, compaction and safe vacuum

id: delta-lake-table-health-gate
namespace: company.team

inputs:
  - id: demo_mode
    type: BOOL
    displayName: Build a throwaway demo table
    description: When true, the flow writes a small Delta table with 40 tiny files
      and works on it. Turn it off to point the flow at your own table.
    defaults: true

  - id: table_uri
    type: STRING
    displayName: Delta table URI
    description: Only used when demo_mode is false. Any location delta-rs can open
      from inside the task container, for example an object store URI. A local
      path on your host is not visible in the container. Credentials are not
      wired in, see "How to extend".
    required: false

  - id: small_file_mb
    type: INT
    displayName: Small file threshold (MB)
    description: A data file smaller than this counts as a small file.
    defaults: 32

  - id: min_small_files
    type: INT
    displayName: Minimum small files before compacting
    description: Compaction is skipped below this count, so a young table is left alone.
    defaults: 10

  - id: small_file_ratio
    type: FLOAT
    displayName: Minimum small file ratio before compacting
    description: Share of data files that must be small, from 0 to 1.
    defaults: 0.5

  - id: target_file_mb
    type: INT
    displayName: Compaction target file size (MB)
    defaults: 128

  - id: retention_hours
    type: INT
    displayName: Vacuum retention (hours)
    description: Files younger than this are never deleted. 168 hours (7 days) is
      the Delta default and the minimum delta-rs accepts. The flow does not
      bypass that check, and rejects lower values before it touches the table.
    defaults: 168
    min: 168

  - id: vacuum_mode
    type: SELECT
    displayName: Vacuum mode
    description: DRY_RUN only lists what vacuum would remove. APPLY deletes files
      that are older than the retention period and no longer referenced by the
      table.
    values:
      - DRY_RUN
      - APPLY
    defaults: DRY_RUN

  - id: notify_slack
    type: BOOL
    displayName: Notify Slack when maintenance changed something
    defaults: false

triggers:
  - id: nightly
    type: io.kestra.plugin.core.trigger.Schedule
    description: Nightly maintenance window. Disabled by default, enable it once the
      table_uri input points at your own table.
    cron: "0 3 * * *"
    disabled: true

tasks:
  - id: work
    type: io.kestra.plugin.core.flow.WorkingDirectory
    description: Keeps the demo table and every step that touches it in one directory.
    tasks:
      - id: seed_demo_table
        type: io.kestra.plugin.scripts.python.Script
        description: Write 40 small appends to a local Delta table, which is how
          streaming and micro-batch jobs fragment a table.
        runIf: "{{ inputs.demo_mode }}"
        taskRunner:
          type: io.kestra.plugin.scripts.runner.docker.Docker
        containerImage: python:3.12-slim
        beforeCommands:
          - pip install -q deltalake==1.6.6 pyarrow==25.0.1 kestra==2.0.0
        script: |
          import pyarrow as pa
          from deltalake import write_deltalake
          from kestra import Kestra

          COMMITS = 40
          for i in range(COMMITS):
              rows = range(i * 100, (i + 1) * 100)
              batch = pa.table({
                  "order_id": list(rows),
                  "amount": [round((r % 50) * 1.5, 2) for r in rows],
              })
              write_deltalake("demo_table", batch, mode="append")

          print(f"Wrote {COMMITS} commits to demo_table")
          Kestra.outputs({"commits": COMMITS})

      - id: assess_table
        type: io.kestra.plugin.scripts.python.Script
        description: Count data files and small files, then decide whether the table
          needs compaction.
        taskRunner:
          type: io.kestra.plugin.scripts.runner.docker.Docker
        containerImage: python:3.12-slim
        beforeCommands:
          - pip install -q deltalake==1.6.6 pyarrow==25.0.1 kestra==2.0.0
        env:
          TABLE_URI: "{{ inputs.demo_mode ? 'demo_table' : inputs.table_uri }}"
          SMALL_FILE_MB: "{{ inputs.small_file_mb }}"
          MIN_SMALL_FILES: "{{ inputs.min_small_files }}"
          SMALL_FILE_RATIO: "{{ inputs.small_file_ratio }}"
        script: |
          import os
          import pyarrow as pa
          from deltalake import DeltaTable
          from kestra import Kestra

          uri = os.environ.get("TABLE_URI", "")
          if not uri:
              raise SystemExit("table_uri is empty. Set it, or keep demo_mode on.")

          small_bytes = float(os.environ["SMALL_FILE_MB"]) * 1024 * 1024
          table = DeltaTable(uri)
          sizes = pa.table(table.get_add_actions(flatten=True)).column("size_bytes").to_pylist()

          small = sum(1 for s in sizes if s < small_bytes)
          ratio = round(small / len(sizes), 3) if sizes else 0.0
          needs = small >= int(os.environ["MIN_SMALL_FILES"]) and ratio >= float(os.environ["SMALL_FILE_RATIO"])

          print(f"version {table.version()}: {len(sizes)} files, {small} small, ratio {ratio}, compaction needed: {needs}")
          Kestra.outputs({
              "version": table.version(),
              "files": len(sizes),
              "small_files": small,
              "small_ratio": ratio,
              "size_mb": round(sum(sizes) / 1048576, 2),
              "needs_compaction": needs,
          })

      - id: compact_small_files
        type: io.kestra.plugin.scripts.python.Script
        description: Rewrite small files into files near the target size. This adds a
          new table version and leaves the old files in place until vacuum
          removes them.
        taskRunner:
          type: io.kestra.plugin.scripts.runner.docker.Docker
        containerImage: python:3.12-slim
        beforeCommands:
          - pip install -q deltalake==1.6.6 pyarrow==25.0.1 kestra==2.0.0
        env:
          TABLE_URI: "{{ inputs.demo_mode ? 'demo_table' : inputs.table_uri }}"
          NEEDS_COMPACTION: "{{ outputs.assess_table.vars.needs_compaction }}"
          TARGET_FILE_MB: "{{ inputs.target_file_mb }}"
        script: |
          import os
          import pyarrow as pa
          from deltalake import DeltaTable
          from kestra import Kestra

          uri = os.environ["TABLE_URI"]
          table = DeltaTable(uri)
          compacted = os.environ["NEEDS_COMPACTION"].lower() == "true"
          removed = added = 0

          if compacted:
              target = int(float(os.environ["TARGET_FILE_MB"]) * 1024 * 1024)
              metrics = table.optimize.compact(target_size=target)
              removed, added = metrics["numFilesRemoved"], metrics["numFilesAdded"]
              table = DeltaTable(uri)
          else:
              print("Below the small file thresholds, nothing to compact.")

          files_after = pa.table(table.get_add_actions(flatten=True)).num_rows
          print(f"compacted={compacted} removed={removed} added={added} files_after={files_after}")
          Kestra.outputs({
              "compacted": compacted,
              "files_removed": removed,
              "files_added": added,
              "files_after": files_after,
              "version": table.version(),
          })

      - id: vacuum_unreferenced
        type: io.kestra.plugin.scripts.python.Script
        description: List files vacuum could remove. Files are only deleted when
          vacuum_mode is APPLY, and never inside the retention period.
        taskRunner:
          type: io.kestra.plugin.scripts.runner.docker.Docker
        containerImage: python:3.12-slim
        beforeCommands:
          - pip install -q deltalake==1.6.6 pyarrow==25.0.1 kestra==2.0.0
        env:
          TABLE_URI: "{{ inputs.demo_mode ? 'demo_table' : inputs.table_uri }}"
          RETENTION_HOURS: "{{ inputs.retention_hours }}"
          VACUUM_MODE: "{{ inputs.vacuum_mode }}"
        script: |
          import os
          from deltalake import DeltaTable
          from kestra import Kestra

          hours = int(os.environ["RETENTION_HOURS"])
          apply_vacuum = os.environ["VACUUM_MODE"] == "APPLY"
          table = DeltaTable(os.environ["TABLE_URI"])

          # Both listings are dry runs. Nothing is deleted by these two calls.
          # The first call raises if hours is below the 168 hour minimum.
          eligible = table.vacuum(retention_hours=hours, dry_run=True)
          unreferenced = table.vacuum(retention_hours=0, dry_run=True, enforce_retention_duration=False)

          deleted = 0
          if apply_vacuum and eligible:
              deleted = len(table.vacuum(retention_hours=hours, dry_run=False))

          print(f"mode={os.environ['VACUUM_MODE']} unreferenced={len(unreferenced)} eligible={len(eligible)} deleted={deleted}")
          Kestra.outputs({
              "unreferenced_files": len(unreferenced),
              "eligible_files": len(eligible),
              "deleted_files": deleted,
          })

  - id: report
    type: io.kestra.plugin.core.log.Log
    message: |
      Delta table: {{ inputs.demo_mode ? 'demo_table (throwaway data)' : inputs.table_uri }}
      Before: {{ outputs.assess_table.vars.files }} files, {{ outputs.assess_table.vars.small_files }} under {{ inputs.small_file_mb }} MB, ratio {{ outputs.assess_table.vars.small_ratio }}.
      Compaction: {{ outputs.compact_small_files.vars.compacted ? 'ran, ' ~ outputs.compact_small_files.vars.files_removed ~ ' files rewritten into ' ~ outputs.compact_small_files.vars.files_added : 'skipped, below thresholds' }}. Files now: {{ outputs.compact_small_files.vars.files_after }}.
      Vacuum ({{ inputs.vacuum_mode }}): {{ outputs.vacuum_unreferenced.vars.unreferenced_files }} unreferenced files on disk, {{ outputs.vacuum_unreferenced.vars.eligible_files }} older than {{ inputs.retention_hours }} hours, {{ outputs.vacuum_unreferenced.vars.deleted_files }} deleted.

  - id: notify
    type: io.kestra.plugin.core.flow.If
    description: Post to Slack only when this run compacted files or deleted files.
    condition: "{{ inputs.notify_slack and
      (outputs.compact_small_files.vars.compacted or
      outputs.vacuum_unreferenced.vars.deleted_files > 0) }}"
    then:
      - id: slack_alert
        type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
        url: "{{ secret('SLACK_WEBHOOK_URL') }}"
        messageText: "Delta maintenance on {{ inputs.demo_mode ? 'demo_table' :
          inputs.table_uri }}: {{ outputs.compact_small_files.vars.files_removed
          }} files compacted, {{ outputs.vacuum_unreferenced.vars.deleted_files
          }} deleted."

errors:
  - id: report_failure
    type: io.kestra.plugin.core.log.Log
    level: ERROR
    message: "Delta table health gate failed in {{ flow.id }}, execution {{
      execution.id }}. Usual causes: table_uri is empty or unreachable, the
      container cannot reach PyPI for the pip install, or retention_hours is
      below 168. Compaction and vacuum are separate table commits, so a failed
      run leaves the table readable."

Streaming and micro-batch jobs write many small files into a Delta table. Reads slow down as the file count grows, and nothing outside a managed platform compacts the table for you. Compaction and vacuum are the two maintenance commands that fix this, and they need different care. Compaction only adds a new table version. Vacuum deletes files, which removes the history that time travel and slow readers depend on.

This flow measures the table first and compacts only when the numbers justify it. It then lists what vacuum could remove, and deletes files only when you ask for APPLY and the files are older than the retention period. It runs with no credentials: in demo mode it builds its own throwaway Delta table.

This blueprint was created by Uttam345.

What you get

A before and after view of the table: file count, small file count, small file ratio, table size.
Compaction that is skipped on healthy tables, so a nightly schedule does not create a new version for nothing.
A vacuum that defaults to DRY_RUN and cannot be pointed below the 168 hour minimum, because the input has a floor of 168 and delta-rs rejects lower values.
A report line per run, plus an optional Slack message when a run compacted or deleted files.

Who it's for

Data engineers who run Delta tables outside Databricks, on object storage, and maintain them by hand.
Teams that want compaction on a schedule but want a dry run before anything is deleted.

Why orchestrate this with Kestra

A cron job running two Python calls has no record of what it changed and no way to stop between measuring and acting. Here each step is its own task with its own logs and outputs. The decision to compact is a visible output of assess_table. The vacuum listing is a separate task, so a failure there leaves the compacted table intact and the run shows exactly which step stopped.

Prerequisites

Docker for the task runner. Each Python task runs in python:3.12-slim.
Network access to PyPI. Every Python task runs pip install for deltalake==1.6.6, pyarrow==25.0.1 and kestra==2.0.0, so a run takes a little longer than the work itself.
No secrets for the demo.
Optional secret SLACK_WEBHOOK_URL, used only when notify_slack is true.

How it works

work (io.kestra.plugin.core.flow.WorkingDirectory) keeps the demo table and every step that touches it in one directory.
seed_demo_table (io.kestra.plugin.scripts.python.Script) runs only in demo mode. It appends 40 small batches with write_deltalake, which leaves 40 tiny data files.
assess_table reads get_add_actions and counts files under small_file_mb. It sets needs_compaction when both min_small_files and small_file_ratio are met.
compact_small_files runs optimize.compact toward target_file_mb when needs_compaction is true, then counts the files again.
vacuum_unreferenced makes two dry runs. One lists files older than retention_hours. The other lists every unreferenced file regardless of age, with the retention check turned off, which is safe only because it deletes nothing. With vacuum_mode: APPLY it then deletes the first list and nothing else.
report (io.kestra.plugin.core.log.Log) prints the before and after summary.
notify (io.kestra.plugin.core.flow.If) posts to Slack through io.kestra.plugin.slack.notifications.SlackIncomingWebhook when notify_slack is true and the run compacted or deleted files.
errors logs the usual causes when any step fails.

Inputs

demo_mode (BOOL, default true): build and use a throwaway table.
table_uri (STRING, optional): your table, used when demo_mode is false. It must be reachable from inside the task container, so a path on your host will not work.
small_file_mb (INT, default 32): files smaller than this are small.
min_small_files (INT, default 10): do not compact below this count.
small_file_ratio (FLOAT, default 0.5): do not compact below this share of small files.
target_file_mb (INT, default 128): compaction target file size.
retention_hours (INT, default 168, minimum 168): files younger than this are never deleted.
vacuum_mode (SELECT, default DRY_RUN): APPLY deletes eligible files.
notify_slack (BOOL, default false): Slack message when a run changed something.

Outputs

Each Python task returns values you can read in later tasks or in the execution view:

{{ outputs.assess_table.vars.small_ratio }} and {{ outputs.assess_table.vars.needs_compaction }}
{{ outputs.compact_small_files.vars.files_removed }} and {{ outputs.compact_small_files.vars.files_after }}
{{ outputs.vacuum_unreferenced.vars.eligible_files }} and {{ outputs.vacuum_unreferenced.vars.deleted_files }}

Quick start

Save the flow and run it with the defaults.
Expect 40 files, 40 small, then compaction rewriting 40 files into 1.
Expect the vacuum line to show 40 unreferenced files and 0 older than 168 hours. That is correct. The 40 old files stay on disk until they pass the retention period, which is what keeps time travel and running readers safe. Nothing is deleted in the demo.
To see a deletion, point table_uri at a table of your own with demo_mode off and set vacuum_mode to APPLY.

How to extend

Open a table on object storage: table_uri accepts any location delta-rs can open. Pass credentials as environment variables on the Python tasks, for example AWS_ACCESS_KEY_ID: "{{ secret('AWS_ACCESS_KEY_ID') }}". They are not wired here because the demo needs no secrets.
Limit compaction to recent partitions: optimize.compact accepts partition_filters.
Track fragmentation over time: write small_ratio to the KV store with io.kestra.plugin.core.kv.Set and compare it on the next run.
Enable the nightly trigger once table_uri points at your own table. Run it outside your heaviest write window.

Compact small files in a Delta Lake table and vacuum it safely

What you get

Who it's for

Why orchestrate this with Kestra

Prerequisites

How it works

Inputs

Outputs

Quick start

How to extend

Links

Tasks

Orchestrate with Kestra

Share this Blueprint

Create a Blueprint

New to Kestra?

Compact small files in a Delta Lake table and vacuum it safely

What you get

Who it's for

Why orchestrate this with Kestra

Prerequisites

How it works

Inputs

Outputs

Quick start

How to extend

Links

Tasks

Orchestrate with Kestra

Share this Blueprint

Create a Blueprint

More Data Blueprints

Flink Job Restart Loop Sentinel

Flink SQL Pipeline Deploy Gate

Firecrawl Web Scraping Job Error and Stagnation Sentinel

Flink Stateful Job Savepoint and Restart Drill

Fluentd Forward Protocol ACK Timeout Sentinel

Trigger multiple Fivetran syncs in parallel, then run a dbt Cloud job

New to Kestra?