Delta Lake Partition Compactor and Safe Vacuum Gate

id: delta-lake-uncompacted-small-files-compactor
namespace: company.data

description: |
  Inspect Delta Lake table partition files via DuckDB and Python, calculate small-file fragmentation ratios,
  detect read query latency degradation, trigger bin-packing and Z-Order compaction when small files exceed 40%,
  and enforce safe vacuum retention gates to protect concurrent readers.

inputs:
  - id: demo_mode
    type: BOOL
    displayName: Generate Synthetic Fragmented Table
    description: When true, builds a partitioned Delta Lake table with fragmented
      micro-batches to demonstrate partition inspection, compaction, and
      vacuuming.
    defaults: true

  - id: table_uri
    type: STRING
    displayName: Delta Table URI
    description: Storage path or cloud object store URI (e.g., s3://bucket/events,
      az://container/events, or local path) to the Delta Lake table.
    defaults: "demo_delta_table"

  - id: small_file_threshold_mb
    type: INT
    displayName: Small File Threshold (MB)
    description: Files smaller than this size in megabytes are classified as
      uncompacted small files.
    defaults: 128

  - id: compaction_ratio_threshold_pct
    type: FLOAT
    displayName: Compaction Small File Ratio Threshold (%)
    description: Minimum percentage of small files across partitions required to
      trigger bin-packing compaction (default 40.0%).
    defaults: 40.0

  - id: target_file_size_mb
    type: INT
    displayName: Target Compacted File Size (MB)
    description: Target Parquet file size in megabytes for bin-packed data files
      during compaction.
    defaults: 128

  - id: z_order_columns
    type: ARRAY
    itemType: STRING
    displayName: Z-Order Optimization Columns
    description: Columns to apply multidimensional Z-Ordering clustering on during
      compaction. If empty, standard bin-packing compaction is executed.
    defaults:
      - "customer_id"

  - id: vacuum_retention_hours
    type: INT
    displayName: Vacuum Retention Period (Hours)
    description: Minimum retention duration in hours before tombstoned data files
      can be purged. Hard minimum is 168 hours (7 days) to protect active and
      concurrent readers.
    defaults: 168
    min: 168

  - id: vacuum_dry_run
    type: BOOL
    displayName: Vacuum Dry Run Mode
    description: When true, lists candidate tombstoned Parquet files eligible for
      vacuuming without physically deleting them from storage.
    defaults: true

  - id: latency_sla_ms
    type: FLOAT
    displayName: Read Latency Baseline SLA (ms)
    description: Baseline read query latency threshold in milliseconds used to
      detect performance degradation caused by partition file fragmentation.
    defaults: 150.0

  - id: notify_slack
    type: BOOL
    displayName: Send Slack Notification
    description: Toggle whether compaction and vacuum audit summaries are sent to Slack.
    defaults: false

triggers:
  - id: nightly_compaction_schedule
    type: io.kestra.plugin.core.trigger.Schedule
    description: Scheduled nightly maintenance window to audit partition health and
      compact uncompacted Delta Lake tables.
    cron: "0 2 * * *"
    disabled: true

  - id: on_demand_compaction_webhook
    type: io.kestra.plugin.core.trigger.Webhook
    description: Authenticated HTTP endpoint invoked by streaming ingestion engines
      (Flink, Spark, Kafka Connect) upon completing batch commits.
    key: "{{ secret('DELTA_COMPACTOR_WEBHOOK_KEY') }}"

tasks:
  - id: validate_safety_guards
    type: io.kestra.plugin.core.execution.Assert
    description: Assert configuration parameters meet operational safety thresholds
      and protect Delta Lake table reader integrity.
    conditions:
      - "{{ inputs.vacuum_retention_hours >= 168 }}"
      - "{{ inputs.small_file_threshold_mb > 0 }}"
      - "{{ inputs.compaction_ratio_threshold_pct > 0.0 and
        inputs.compaction_ratio_threshold_pct <= 100.0 }}"
      - "{{ inputs.target_file_size_mb > 0 }}"
    errorMessage: Invalid configuration parameters. Vacuum retention must be at
      least 168 hours (7 days), and file size thresholds must be positive
      values.

  - id: table_workspace
    type: io.kestra.plugin.core.flow.WorkingDirectory
    description: Isolates Delta Lake table operations in a shared workspace across
      inspection, compaction, and vacuum tasks.
    tasks:
      - id: seed_synthetic_microbatches
        type: io.kestra.plugin.scripts.python.Script
        description: Simulates continuous micro-batch ingestion by writing 30 small
          commits partitioned by date into the Delta table.
        runIf: "{{ inputs.demo_mode }}"
        taskRunner:
          type: io.kestra.plugin.scripts.runner.docker.Docker
        containerImage: python:3.11-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

          partition_dates = ["2026-10-17", "2026-10-18", "2026-10-19"]
          commits_count = 30
          rows_per_commit = 50

          for i in range(commits_count):
              part_date = partition_dates[i % len(partition_dates)]
              batch = pa.table({
                  "event_id": [f"EVT_{i}_{r:05d}" for r in range(rows_per_commit)],
                  "customer_id": [f"CUST_{((i * 10) + r) % 40:04d}" for r in range(rows_per_commit)],
                  "event_type": ["click" if r % 3 == 0 else ("purchase" if r % 3 == 1 else "search") for r in range(rows_per_commit)],
                  "amount": [round(((r % 25) + 1) * 3.75, 2) for r in range(rows_per_commit)],
                  "event_date": [part_date] * rows_per_commit
              })
              write_deltalake("demo_delta_table", batch, partition_by=["event_date"], mode="append")

          print(f"Successfully generated {commits_count} micro-batches in demo_delta_table across {len(partition_dates)} partitions.")
          Kestra.outputs({
              "commits_seeded": commits_count,
              "rows_seeded": commits_count * rows_per_commit
          })

      - id: inspect_partitions_and_latency
        type: io.kestra.plugin.scripts.python.Script
        description: Audits Delta Lake partition metadata, calculates small-file
          fragmentation ratios, and benchmarks read query latency using DuckDB.
        taskRunner:
          type: io.kestra.plugin.scripts.runner.docker.Docker
        containerImage: python:3.11-slim
        beforeCommands:
          - pip install -q deltalake==1.6.6 duckdb==1.5.6 pyarrow==25.0.1
            kestra==2.0.0
        env:
          TABLE_URI: "{{ inputs.demo_mode ? 'demo_delta_table' : inputs.table_uri }}"
          SMALL_FILE_MB: "{{ inputs.small_file_threshold_mb }}"
          COMPACTION_RATIO_PCT: "{{ inputs.compaction_ratio_threshold_pct }}"
          LATENCY_SLA_MS: "{{ inputs.latency_sla_ms }}"
        script: |
          import os
          import time
          import pyarrow as pa
          from deltalake import DeltaTable
          import duckdb
          from kestra import Kestra

          table_uri = os.environ.get("TABLE_URI", "demo_delta_table")
          small_file_mb = float(os.environ.get("SMALL_FILE_MB", "128"))
          small_file_bytes = int(small_file_mb * 1024 * 1024)
          compaction_ratio_pct = float(os.environ.get("COMPACTION_RATIO_PCT", "40.0"))
          latency_sla_ms = float(os.environ.get("LATENCY_SLA_MS", "150.0"))

          dt = DeltaTable(table_uri)
          version = dt.version()
          add_actions = pa.table(dt.get_add_actions(flatten=True)).to_pylist()

          total_files = len(add_actions)
          total_bytes = sum(a["size_bytes"] for a in add_actions)

          # Aggregate partition-level distribution
          partition_stats = {}
          total_small_files = 0

          for action in add_actions:
              size = action.get("size_bytes", 0)
              # Discover partition keys dynamically
              part_keys = [k for k in action.keys() if k.startswith("partition.")]
              if part_keys:
                  part_identifier = "/".join(f"{k.replace('partition.', '')}={action[k]}" for k in sorted(part_keys))
              else:
                  part_identifier = "unpartitioned"

              if part_identifier not in partition_stats:
                  partition_stats[part_identifier] = {
                      "partition": part_identifier,
                      "file_count": 0,
                      "total_bytes": 0,
                      "small_files_count": 0
                  }

              p = partition_stats[part_identifier]
              p["file_count"] += 1
              p["total_bytes"] += size
              if size < small_file_bytes:
                  p["small_files_count"] += 1
                  total_small_files += 1

          partition_summary = []
          for p_id, stats in sorted(partition_stats.items()):
              f_cnt = stats["file_count"]
              sm_cnt = stats["small_files_count"]
              tot_b = stats["total_bytes"]
              avg_b = tot_b / f_cnt if f_cnt else 0
              ratio = round((sm_cnt / f_cnt) * 100.0, 2) if f_cnt else 0.0
              partition_summary.append({
                  "partition": p_id,
                  "file_count": f_cnt,
                  "small_files_count": sm_cnt,
                  "small_file_ratio_pct": ratio,
                  "avg_file_size_mb": round(avg_b / (1024.0 * 1024.0), 3),
                  "total_size_mb": round(tot_b / (1024.0 * 1024.0), 3)
              })

          overall_small_file_ratio_pct = round((total_small_files / total_files) * 100.0, 2) if total_files else 0.0
          avg_file_size_overall_mb = round((total_bytes / total_files) / (1024.0 * 1024.0), 3) if total_files else 0.0
          total_size_mb = round(total_bytes / (1024.0 * 1024.0), 3)

          # Benchmark read query scan latency using DuckDB
          con = duckdb.connect()
          normalized_uri = table_uri.replace("\\", "/")
          scan_sql = f"SELECT count(*), sum(amount) FROM delta_scan('{normalized_uri}')"

          start_time = time.perf_counter()
          con.execute(scan_sql).fetchall()
          read_latency_pre_ms = round((time.perf_counter() - start_time) * 1000.0, 2)

          latency_regression_detected = read_latency_pre_ms > latency_sla_ms
          compaction_required = overall_small_file_ratio_pct >= compaction_ratio_pct

          print(f"Table '{table_uri}' Version {version}: {total_files} active files, {total_small_files} small files (< {small_file_mb} MB).")
          print(f"Small file ratio: {overall_small_file_ratio_pct}% (Threshold: {compaction_ratio_pct}%). Compaction required: {compaction_required}.")
          print(f"Read latency: {read_latency_pre_ms} ms (SLA: {latency_sla_ms} ms). Regression detected: {latency_regression_detected}.")

          Kestra.outputs({
              "table_version": version,
              "total_files": total_files,
              "total_small_files": total_small_files,
              "small_file_ratio_pct": overall_small_file_ratio_pct,
              "avg_file_size_mb": avg_file_size_overall_mb,
              "total_size_mb": total_size_mb,
              "read_latency_pre_ms": read_latency_pre_ms,
              "latency_regression_detected": latency_regression_detected,
              "compaction_required": compaction_required,
              "partition_metrics": partition_summary
          })

      - id: compact_small_files
        type: io.kestra.plugin.scripts.python.Script
        description: Rewrites fragmented small files into target-sized Parquet files
          using bin-packing and Z-Ordering clustering.
        taskRunner:
          type: io.kestra.plugin.scripts.runner.docker.Docker
        containerImage: python:3.11-slim
        beforeCommands:
          - pip install -q deltalake==1.6.6 duckdb==1.5.6 pyarrow==25.0.1
            kestra==2.0.0
        env:
          TABLE_URI: "{{ inputs.demo_mode ? 'demo_delta_table' : inputs.table_uri }}"
          COMPACTION_REQUIRED: "{{
            outputs.inspect_partitions_and_latency.vars.compaction_required }}"
          TARGET_FILE_MB: "{{ inputs.target_file_size_mb }}"
          Z_ORDER_COLUMNS: "{{ inputs.z_order_columns | toJson }}"
          READ_LATENCY_PRE_MS: "{{
            outputs.inspect_partitions_and_latency.vars.read_latency_pre_ms }}"
        script: |
          import os
          import json
          import time
          import pyarrow as pa
          from deltalake import DeltaTable
          import duckdb
          from kestra import Kestra

          table_uri = os.environ.get("TABLE_URI", "demo_delta_table")
          compaction_required = os.environ.get("COMPACTION_REQUIRED", "false").lower() == "true"
          target_file_mb = float(os.environ.get("TARGET_FILE_MB", "128"))
          target_bytes = int(target_file_mb * 1024 * 1024)
          z_order_cols_raw = os.environ.get("Z_ORDER_COLUMNS", "[]")
          read_latency_pre_ms = float(os.environ.get("READ_LATENCY_PRE_MS", "0.0"))

          try:
              z_order_columns = json.loads(z_order_cols_raw) if z_order_cols_raw else []
          except Exception:
              z_order_columns = []

          dt = DeltaTable(table_uri)
          version_before = dt.version()
          add_actions_before = pa.table(dt.get_add_actions(flatten=True)).to_pylist()
          files_before = len(add_actions_before)

          compacted = False
          files_removed = 0
          files_added = 0
          files_reduced = 0
          bytes_compacted = 0
          read_latency_post_ms = read_latency_pre_ms
          latency_speedup_pct = 0.0

          if compaction_required:
              print(f"Triggering compaction on '{table_uri}' (target size: {target_file_mb} MB)...")
              if z_order_columns:
                  print(f"Applying multidimensional Z-Ordering clustering on columns: {z_order_columns}")
                  metrics = dt.optimize.z_order(columns=z_order_columns, target_size=target_bytes)
              else:
                  print("Applying bin-packing compaction...")
                  metrics = dt.optimize.compact(target_size=target_bytes)

              compacted = True
              files_removed = metrics.get("numFilesRemoved", 0)
              files_added = metrics.get("numFilesAdded", 0)
              files_reduced = files_removed - files_added

              removed_details = metrics.get("filesRemoved", {})
              if isinstance(removed_details, str):
                  try:
                      removed_details = json.loads(removed_details)
                  except Exception:
                      removed_details = {}
              bytes_compacted = removed_details.get("totalSize", 0)

              # Refresh DeltaTable reference
              dt = DeltaTable(table_uri)

              # Post-compaction read latency benchmark
              con = duckdb.connect()
              normalized_uri = table_uri.replace("\\", "/")
              scan_sql = f"SELECT count(*), sum(amount) FROM delta_scan('{normalized_uri}')"

              start_time = time.perf_counter()
              con.execute(scan_sql).fetchall()
              read_latency_post_ms = round((time.perf_counter() - start_time) * 1000.0, 2)

              if read_latency_pre_ms > 0:
                  latency_speedup_pct = round(((read_latency_pre_ms - read_latency_post_ms) / read_latency_pre_ms) * 100.0, 2)

              print(f"Compaction completed. Files removed: {files_removed}, Files added: {files_added}, Files reduced: {files_reduced}.")
              print(f"Bytes compacted: {bytes_compacted} bytes ({round(bytes_compacted / (1024 * 1024), 2)} MB).")
              print(f"Read latency: {read_latency_pre_ms} ms -> {read_latency_post_ms} ms (Speedup: {latency_speedup_pct}%).")
          else:
              print("Small file ratio is below threshold. Compaction skipped.")

          add_actions_after = pa.table(dt.get_add_actions(flatten=True)).to_pylist()
          files_after = len(add_actions_after)

          Kestra.outputs({
              "compacted": compacted,
              "version_before": version_before,
              "version_after": dt.version(),
              "files_before": files_before,
              "files_after": files_after,
              "files_removed": files_removed,
              "files_added": files_added,
              "files_reduced": files_reduced,
              "bytes_compacted": bytes_compacted,
              "bytes_compacted_mb": round(bytes_compacted / (1024.0 * 1024.0), 3),
              "read_latency_post_ms": read_latency_post_ms,
              "latency_speedup_pct": latency_speedup_pct
          })

      - id: execute_safe_vacuum
        type: io.kestra.plugin.scripts.python.Script
        description: Purges unreferenced tombstoned Parquet files while enforcing the
          168-hour retention guard to protect active readers.
        taskRunner:
          type: io.kestra.plugin.scripts.runner.docker.Docker
        containerImage: python:3.11-slim
        beforeCommands:
          - pip install -q deltalake==1.6.6 pyarrow==25.0.1 kestra==2.0.0
        env:
          TABLE_URI: "{{ inputs.demo_mode ? 'demo_delta_table' : inputs.table_uri }}"
          RETENTION_HOURS: "{{ inputs.vacuum_retention_hours }}"
          DRY_RUN: "{{ inputs.vacuum_dry_run }}"
        script: |
          import os
          from deltalake import DeltaTable
          from kestra import Kestra

          table_uri = os.environ.get("TABLE_URI", "demo_delta_table")
          retention_hours = int(os.environ.get("RETENTION_HOURS", "168"))
          dry_run = os.environ.get("DRY_RUN", "true").lower() == "true"

          if retention_hours < 168:
              raise ValueError(f"Retention safety violation: retention_hours ({retention_hours}) is lower than the mandatory 168-hour limit.")

          dt = DeltaTable(table_uri)

          # Execute vacuum dry run first to identify candidates
          candidate_files = dt.vacuum(retention_hours=retention_hours, dry_run=True)
          candidate_count = len(candidate_files)

          deleted_count = 0
          if not dry_run and candidate_count > 0:
              purged = dt.vacuum(retention_hours=retention_hours, dry_run=False)
              deleted_count = len(purged)
              print(f"Safe VACUUM applied: purged {deleted_count} tombstoned files older than {retention_hours} hours.")
          else:
              mode_str = "DRY RUN" if dry_run else "APPLY (no candidates older than retention threshold)"
              print(f"Safe VACUUM {mode_str}: {candidate_count} files eligible for retention purge.")

          Kestra.outputs({
              "vacuum_retention_hours": retention_hours,
              "vacuum_dry_run": dry_run,
              "candidate_files_count": candidate_count,
              "deleted_files_count": deleted_count
          })

  - id: persist_compaction_metrics_to_kv
    type: io.kestra.plugin.core.kv.Set
    description: Persists compaction metrics and partition health telemetry to the
      Kestra KV store.
    key: "delta_compaction_{{ inputs.demo_mode ? 'demo_delta_table' :
      (inputs.table_uri | slugify) }}"
    kvType: JSON
    value: |
      {
        "table_uri": "{{ inputs.demo_mode ? 'demo_delta_table' : inputs.table_uri }}",
        "evaluated_at": "{{ now() }}",
        "execution_id": "{{ execution.id }}",
        "small_file_threshold_mb": {{ inputs.small_file_threshold_mb }},
        "compaction_ratio_threshold_pct": {{ inputs.compaction_ratio_threshold_pct }},
        "total_files_before": {{ outputs.inspect_partitions_and_latency.vars.total_files }},
        "small_files_before": {{ outputs.inspect_partitions_and_latency.vars.total_small_files }},
        "small_file_ratio_pct": {{ outputs.inspect_partitions_and_latency.vars.small_file_ratio_pct }},
        "compaction_executed": {{ outputs.compact_small_files.vars.compacted }},
        "files_after": {{ outputs.compact_small_files.vars.files_after }},
        "files_reduced": {{ outputs.compact_small_files.vars.files_reduced }},
        "bytes_compacted": {{ outputs.compact_small_files.vars.bytes_compacted }},
        "bytes_compacted_mb": {{ outputs.compact_small_files.vars.bytes_compacted_mb }},
        "read_latency_pre_ms": {{ outputs.inspect_partitions_and_latency.vars.read_latency_pre_ms }},
        "read_latency_post_ms": {{ outputs.compact_small_files.vars.read_latency_post_ms }},
        "latency_speedup_pct": {{ outputs.compact_small_files.vars.latency_speedup_pct }},
        "latency_regression_detected": {{ outputs.inspect_partitions_and_latency.vars.latency_regression_detected }},
        "vacuum_retention_hours": {{ inputs.vacuum_retention_hours }},
        "vacuum_dry_run": {{ inputs.vacuum_dry_run }},
        "vacuum_candidate_files": {{ outputs.execute_safe_vacuum.vars.candidate_files_count }},
        "vacuum_deleted_files": {{ outputs.execute_safe_vacuum.vars.deleted_files_count }}
      }

  - id: log_compaction_summary
    type: io.kestra.plugin.core.log.Log
    description: Logs summary of compaction, latency metrics, and safe vacuum results.
    message: |
      Delta Lake Compaction & Vacuum Gate Summary:
      Table: {{ inputs.demo_mode ? 'demo_delta_table' : inputs.table_uri }}
      Initial State: {{ outputs.inspect_partitions_and_latency.vars.total_files }} files ({{ outputs.inspect_partitions_and_latency.vars.total_small_files }} small files, ratio: {{ outputs.inspect_partitions_and_latency.vars.small_file_ratio_pct }}%).
      Compaction Status: {{ outputs.compact_small_files.vars.compacted ? 'EXECUTED (' ~ outputs.compact_small_files.vars.files_reduced ~ ' files reduced, ' ~ outputs.compact_small_files.vars.bytes_compacted_mb ~ ' MB compacted)' : 'SKIPPED (below 40% threshold)' }}.
      Read Latency: {{ outputs.inspect_partitions_and_latency.vars.read_latency_pre_ms }} ms -> {{ outputs.compact_small_files.vars.read_latency_post_ms }} ms (speedup: {{ outputs.compact_small_files.vars.latency_speedup_pct }}%).
      Safe Vacuum ({{ inputs.vacuum_dry_run ? 'DRY RUN' : 'APPLY' }}): {{ outputs.execute_safe_vacuum.vars.candidate_files_count }} candidates, {{ outputs.execute_safe_vacuum.vars.deleted_files_count }} files purged (retention: {{ inputs.vacuum_retention_hours }}h).

  - id: notify_slack_summary
    type: io.kestra.plugin.core.flow.If
    description: Posts detailed compaction audit report to Slack if Slack
      notifications are enabled.
    condition: "{{ inputs.notify_slack == true }}"
    then:
      - id: send_slack_compaction_report
        type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
        description: Sends formatted Slack notification detailing compaction gains and
          vacuum results.
        url: "{{ secret('SLACK_WEBHOOK_URL') }}"
        payload: |
          {
            "attachments": [
              {
                "color": "{{ outputs.compact_small_files.vars.compacted ? '#36a64f' : '#439FE0' }}",
                "title": "Delta Lake Small Files Compactor: {{ outputs.compact_small_files.vars.compacted ? 'Compaction Executed' : 'Table Healthy' }}",
                "text": "*Table:* `{{ inputs.demo_mode ? 'demo_delta_table' : inputs.table_uri }}`\n*Execution:* `{{ execution.id }}`\n*Initial Files:* {{ outputs.inspect_partitions_and_latency.vars.total_files }} ({{ outputs.inspect_partitions_and_latency.vars.total_small_files }} small files, *{{ outputs.inspect_partitions_and_latency.vars.small_file_ratio_pct }}%* ratio)\n*Compacted Files:* {{ outputs.compact_small_files.vars.files_reduced }} files reduced ({{ outputs.compact_small_files.vars.bytes_compacted_mb }} MB rewritten)\n*Read Latency:* {{ outputs.inspect_partitions_and_latency.vars.read_latency_pre_ms }} ms → {{ outputs.compact_small_files.vars.read_latency_post_ms }} ms (*{{ outputs.compact_small_files.vars.latency_speedup_pct }}%* speedup)\n*Safe Vacuum:* {{ outputs.execute_safe_vacuum.vars.deleted_files_count }} purged, {{ outputs.execute_safe_vacuum.vars.candidate_files_count }} candidates (retention: {{ inputs.vacuum_retention_hours }}h, dry-run: {{ inputs.vacuum_dry_run }})"
              }
            ]
          }

errors:
  - id: log_compaction_failure
    type: io.kestra.plugin.core.log.Log
    level: ERROR
    message: "Delta Lake small files compactor failed on execution {{ execution.id
      }} for table {{ inputs.table_uri }}."

  - id: notify_slack_failure
    type: io.kestra.plugin.core.flow.If
    description: Dispatches Slack alert on workflow execution failure if
      notifications are enabled.
    condition: "{{ inputs.notify_slack == true }}"
    then:
      - id: send_slack_failure_alert
        type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
        description: Alerts engineering team of compaction or vacuum pipeline failure.
        url: "{{ secret('SLACK_WEBHOOK_URL') }}"
        payload: |
          {
            "attachments": [
              {
                "color": "danger",
                "title": ":x: Delta Lake Compaction Pipeline Failed",
                "text": "Execution `{{ execution.id }}` failed for Delta table `{{ inputs.table_uri }}`. Inspect execution task logs for details."
              }
            ]
          }

Continuous streaming and micro-batch ingestion create thousands of small Parquet files in Delta Lake tables. As small files accumulate across table partitions, metadata overhead surges, object store request costs rise, and query scan latency degrades severely. External compaction requires careful coordination: merging small files must balance bin-packing target sizes with multidimensional Z-Ordering clustering, while tombstone vacuuming must enforce strict retention windows to prevent corrupting concurrent readers.

This blueprint implements an automated Day 2 operations maintenance gate for Delta Lake tables:

Partition-Level Inspection: Queries Delta transaction log metadata via deltalake and DuckDB to measure average file sizes and small-file fragmentation ratios per partition.
Query Latency Benchmarking: Executes partition scan queries through DuckDB to detect read performance regressions against configured SLA thresholds.
Dynamic Compaction Gate: Triggers bin-packing compaction and multidimensional Z-Ordering when small files exceed 40% of active partition files, rewriting micro-batches into 128 MB Parquet files.
Safe Vacuum Enforcement: Runs VACUUM with a hard safety guard enforcing at least 168 hours (7 days) of retention, preventing premature deletion of historical files needed by active readers and time-travel queries.
Stateful KV Telemetry: Persists compaction metrics (files reduced, bytes rewritten, latency speedup) into the Kestra KV store and dispatches notifications to Slack.

How it works

Configuration Validation: An Assert task validates that safety parameters are respected, strictly requiring vacuum retention to be at least 168 hours.
Synthetic Data Seeding: If demo_mode is enabled, the flow writes 30 small micro-batch appends across partitioned dates to simulate streaming fragmentation.
Partition & Latency Audit: A Python task scans transaction log commit actions, calculates small-file percentages across partitions, and benchmarks DuckDB read scan latency against SLA targets.
Bin-Packing & Z-Order Compaction: If small files exceed the 40% threshold, the compaction engine executes OPTIMIZE with Z-Ordering clustering, measures rewritten bytes and reduced files, and benchmarks post-compaction read speedup.
Safe Vacuum Execution: Delta-rs inspects tombstoned files and deletes only unreferenced Parquet files that have exceeded the 168-hour retention window.
KV Telemetry & Slack Reporting: A kv.Set task records compaction metrics into Kestra KV under delta_compaction_{table_identifier}, logs human-readable summaries, and alerts team channels on Slack.

What you get

Automated detection of partition small-file fragmentation from streaming ingest.
Read query latency benchmarking before and after bin-packing compaction.
Safe Parquet file bin-packing into 128 MB targets with Z-Ordering clustering.
Enforced 168-hour retention guard protecting concurrent readers from time-travel corruption.
Historical compaction statistics stored natively in Kestra namespace KV store.
Formatted Slack alerts reporting files reduced, storage compacted, and latency speedup.

Who it's for

Data Platform Engineers operating high-frequency streaming ingest into Delta Lake.
Lakehouse Administrators looking to automate partition compaction and vacuuming.
FinOps & Cloud Infrastructure Teams reducing cloud object storage S3/GCS API request fees.

Why orchestrate this with Kestra

Serverless & Lightweight: Executes Delta Lake optimization and DuckDB queries in isolated Docker containers without requiring persistent Spark clusters or Databricks runtimes.
Deterministic Safety Guards: Prevents operational data corruption by asserting vacuum retention limits before touching storage.
Integrated KV Store: Retains historical compaction telemetry across executions directly in Kestra without external database setup.
Event-Driven & Scheduled Operations: Run automatically on a nightly schedule or trigger instantaneously via authenticated webhooks after batch ingest completes.

Prerequisites

Docker runner configured on Kestra workers to execute containerized Python tasks.
Read and write access to the Delta Lake storage URI (local filesystem or cloud object storage).
Slack Incoming Webhook configured in Kestra secrets (if Slack alerts are enabled).

Secrets

SLACK_WEBHOOK_URL: Slack Incoming Webhook URL for compaction audit summaries.
DELTA_COMPACTOR_WEBHOOK_KEY: Secret authentication key for triggering the compaction webhook.

Quick start

Configure SLACK_WEBHOOK_URL and DELTA_COMPACTOR_WEBHOOK_KEY in your Kestra namespace secrets.
Import this blueprint into your Kestra instance.
Run the flow with demo_mode: true to generate synthetic partitioned micro-batches, observe bin-packing compaction, and inspect post-compaction latency improvements.
Switch demo_mode: false and point table_uri to your production Delta Lake table to schedule automated maintenance.

How to extend

Cloud Object Storage Authentication: Pass cloud credentials (AWS_ACCESS_KEY_ID, AWS_SECRET_ACCESS_KEY, AZURE_STORAGE_KEY, GOOGLE_APPLICATION_CREDENTIALS) into task env blocks to manage S3, ADLS Gen2, or GCS Delta tables.
Partition Range Filtering: Scope compaction to specific rolling partition windows (e.g. event_date >= current_date() - interval '7 days') using delta-rs partition filters.
Downstream Data Quality Gates: Add Great Expectations or Soda Core validation tasks following compaction to verify dataset invariants.

Delta Lake Uncompacted Small Files Compactor and Safe Vacuum Gate

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?

Delta Lake Uncompacted Small Files Compactor and Safe Vacuum Gate

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

Airbyte Connection Sync Failure Sentinel

Invoice Math Discrepancy Gate with In-Memory DuckDB Reconciliation

AI Synthetic Relational Data Generator and Referential Integrity Gate

Generate and Execute SQL Queries from Natural Language Using OpenAI GPT-4o

AI-Powered Inventory Demand Forecasting, Auto-Generate Purchase Orders with OpenAI

Autonomous Mathematical Research and Scientific Computation Pipeline

More Cloud Blueprints

Automated AI Security Vulnerability Triage and GitHub Issue Sync

AI Multi-Provider Consensus Gate and Governance Adjudicator

Ansible Scalability, Deploy Configuration Files to Multiple Servers at Scale

AI Prompt Drift and Quality Sentinel

Apache Hudi Commit Metadata Staleness Watchdog

Argo CD Application Sync Degraded Sentinel

New to Kestra?