Apache Airflow orchestrates Apache Iceberg maintenance workflows when the table format itself does not provide scheduling or workflow management. Its DAGs coordinate execution dependencies, retries, alerting, and conditional logic across SQL-compatible engines such as Apache Spark. section 11.1.1
A metadata-aware workflow can inspect an Iceberg table, compact it only when small-file counts exceed a threshold, then expire snapshots and remove orphan files. This reactive sequence avoids relying on a fixed maintenance schedule. section 11.1.1 section 11.1.5
from airflow import DAG
from airflow.operators.bash import BashOperator
from airflow.operators.python import PythonOperator
from datetime import datetime
import subprocess
default_args = {
'start_date': datetime(2023, 1, 1),
'retries': 1,
}
dag = DAG(
'iceberg_compaction_dag',
schedule_interval='@daily',
default_args=default_args,
catchup=False,
)
def should_compact():
result = subprocess.check_output([
"spark-sql ",
"-e ",
"SELECT count(*) FROM mydb.mytable.files WHERE file_size_in_bytes <20000000;"#1
])
return int(result.strip()) >500 #2
check_compaction = PythonOperator(
task_id='check_file_fragmentation',
python_callable=should_compact,
dag=dag,
)
compact_table = BashOperator(
task_id='run_compaction',
bash_command='spark-sql -e "CALL mydb.system.rewrite_data_files(table =>\'mydb.mytable\')"', #3
dag=dag,
)
check_compaction >>compact_table #4
The example defines a daily DAG with retries, checks file fragmentation through Spark SQL, and invokes Iceberg compaction through a Bash operator. section 11.1.5
