Airflow + StarRocks Backfill Patterns
When to Use
- Reprocess historical partitions after fixing a bug in transform logic
- Re-load data after upstream source correction (wrong values corrected in source)
- Load data for new tables from existing S3 history
- Recover from a failed incremental pipeline that missed N days
- Migrate data from one partition scheme to another
Core Backfill Concepts for StarRocks
StarRocks backfill safety requirements:
- INSERT OVERWRITE — replaces partition atomically; safe to re-run for same date
- Broker Load labels — deterministic label per
(table, date) ensures idempotency
- Partition pre-creation — target partitions must exist before INSERT OVERWRITE
- Airflow
max_active_runs=1 — prevents concurrent runs from racing on same partition
Pattern 1: Catchup-Based Backfill
Use Airflow's native catchup mechanism for date-by-date reprocessing:
from airflow.decorators import dag, task
from airflow.providers.mysql.hooks.mysql import MySqlHook
from datetime import datetime, timedelta
@dag(
dag_id="starrocks_orders_backfill",
schedule="@daily",
start_date=datetime(2024, 1, 1), # backfill starts from here
end_date=datetime(2024, 3, 31), # optional: stop backfill here
catchup=True, # process all missed intervals
max_active_runs=1, # one partition at a time
tags=["starrocks", "backfill"],
default_args={
"retries": 3,
"retry_delay": timedelta(minutes=5),
},
)
def orders_backfill():
@task
def ensure_partition(ds: str = None, db: str = "sales"):
hook = MySqlHook(mysql_conn_id="starrocks_prod")
p_name = f"p{ds.replace('-', '')}"
p_start = ds
p_end = (datetime.strptime(ds, "%Y-%m-%d") + timedelta(days=1)).strftime("%Y-%m-%d")
try:
hook.run(f"""
ALTER TABLE {db}.orders
ADD PARTITION {p_name}
VALUES [('{p_start}'), ('{p_end}'))
""")
except Exception as e:
if "already exists" in str(e).lower() or "Duplicate" in str(e):
pass # partition exists — fine
else:
raise
@task
def load_partition(ds: str = None, db: str = "sales"):
hook = MySqlHook(mysql_conn_id="starrocks_prod")
label = f"backfill_orders_{ds.replace('-', '')}"
# Check if already completed
rows = hook.get_records(
f"SHOW LOAD FROM {db} WHERE LABEL = '{label}'"
)
if rows and rows[0][2] == "FINISHED":
print(f"Partition {ds} already loaded — skipping")
return
hook.run(f"""
LOAD LABEL {db}.{label} (
DATA INFILE("s3a://datalake/orders/dt={ds}/*.parquet")
INTO TABLE orders
FORMAT AS "parquet"
)
WITH BROKER (
"aws.s3.use_instance_profile" = "true",
"aws.s3.region" = "us-east-1"
)
PROPERTIES (
"timeout" = "3600",
"max_filter_ratio" = "0.01",
"load_parallelism" = "4"
)
""")
# Poll for completion
import time
for _ in range(120): # wait up to 60 min
rows = hook.get_records(
f"SHOW LOAD FROM {db} WHERE LABEL = '{label}'"
)
if rows:
state = rows[0][2]
if state == "FINISHED":
return
elif state == "CANCELLED":
raise RuntimeError(f"Load CANCELLED for {ds}: {rows[0][7]}")
time.sleep(30)
raise TimeoutError(f"Load for {ds} timed out")
@task
def analyze_partition(ds: str = None, db: str = "sales"):
hook = MySqlHook(mysql_conn_id="starrocks_prod")
p_name = f"p{ds.replace('-', '')}"
hook.run(f"""
ANALYZE TABLE {db}.orders PARTITION ({p_name}) WITH ASYNC MODE
""")
partition = ensure_partition()
load = load_partition()
analyze = analyze_partition()
partition >> load >> analyze
dag = orders_backfill()
Pattern 2: Programmatic Date Range Backfill DAG
Manually triggered backfill for arbitrary date ranges:
from airflow.decorators import dag, task
from airflow.providers.mysql.hooks.mysql import MySqlHook
from datetime import datetime, timedelta
@dag(
dag_id="starrocks_range_backfill",
schedule=None, # triggered manually
start_date=datetime(2024, 1, 1),
params={
"start_date": "2024-01-01",
"end_date": "2024-03-31",
"table": "orders",
"db": "sales",
"dry_run": False,
},
tags=["starrocks", "backfill"],
max_active_runs=1,
)
def range_backfill():
@task
def generate_date_range(params: dict = None) -> list[str]:
start = datetime.strptime(params["start_date"], "%Y-%m-%d")
end = datetime.strptime(params["end_date"], "%Y-%m-%d")
dates = []
cur = start
while cur <= end:
dates.append(cur.strftime("%Y-%m-%d"))
cur += timedelta(days=1)
print(f"Generated {len(dates)} dates: {dates[0]} → {dates[-1]}")
return dates
@task
def check_completed_partitions(dates: list[str], params: dict = None) -> list[str]:
"""Return only dates that need processing (not yet loaded)."""
hook = MySqlHook(mysql_conn_id="starrocks_prod")
db = params["db"]
table = params["table"]
pending = []
for ds in dates:
label = f"backfill_{table}_{ds.replace('-', '')}"
rows = hook.get_records(
f"SHOW LOAD FROM {db} WHERE LABEL = '{label}'"
)
if rows and rows[0][2] == "FINISHED":
print(f" SKIP {ds} — already loaded")
else:
pending.append(ds)
print(f" PENDING {ds}")
print(f"Pending dates: {len(pending)}/{len(dates)}")
return pending
@task
def backfill_date_batch(pending_dates: list[str], params: dict = None):
"""Process pending dates sequentially."""
import time
hook = MySqlHook(mysql_conn_id="starrocks_prod")
db = params["db"]
table = params["table"]
dry_run = params.get("dry_run", False)
results = []
for ds in pending_dates:
print(f"\n--- Processing {ds} ---")
if dry_run:
print(f"DRY RUN: would load {table} for {ds}")
results.append({"ds": ds, "status": "dry_run"})
continue
# Ensure partition exists
p_name = f"p{ds.replace('-', '')}"
p_end = (datetime.strptime(ds, "%Y-%m-%d") + timedelta(days=1)).strftime("%Y-%m-%d")
try:
hook.run(f"""
ALTER TABLE {db}.{table}
ADD PARTITION {p_name} VALUES [('{ds}'), ('{p_end}'))
""")
except Exception as e:
if "already exists" not in str(e).lower():
raise
# Trigger load
label = f"backfill_{table}_{ds.replace('-', '')}"
hook.run(f"""
LOAD LABEL {db}.{label} (
DATA INFILE("s3a://datalake/{table}/dt={ds}/*.parquet")
INTO TABLE {table} FORMAT AS "parquet"
)
WITH BROKER (
"aws.s3.use_instance_profile" = "true",
"aws.s3.region" = "us-east-1"
)
PROPERTIES ("timeout" = "3600", "max_filter_ratio" = "0.01")
""")
# Poll
for _ in range(120):
rows = hook.get_records(
f"SHOW LOAD FROM {db} WHERE LABEL = '{label}'"
)
if rows:
state = rows[0][2]
if state == "FINISHED":
print(f" ✓ {ds} loaded")
results.append({"ds": ds, "status": "ok"})
break
elif state == "CANCELLED":
raise RuntimeError(f"Load CANCELLED for {ds}: {rows[0][7]}")
time.sleep(30)
else:
raise TimeoutError(f"Load timed out for {ds}")
return results
dates = generate_date_range()
pending = check_completed_partitions(dates)
backfill_date_batch(pending)
dag = range_backfill()
Pattern 3: INSERT OVERWRITE Backfill (Transform Re-run)
When the source data is in StarRocks (not S3), re-apply the transform:
@task
def recompute_partition(ds: str = None, db: str = "sales"):
"""Recompute a partition using INSERT OVERWRITE — idempotent."""
hook = MySqlHook(mysql_conn_id="starrocks_prod")
# INSERT OVERWRITE atomically replaces the partition
hook.run(f"""
INSERT OVERWRITE {db}.orders_daily PARTITION (dt='{ds}')
SELECT
DATE(created_at) AS dt,
customer_id,
COUNT(*) AS order_count,
SUM(amount) AS total_revenue,
AVG(amount) AS avg_order_value,
COUNT(DISTINCT status) AS status_variety
FROM {db}.orders
WHERE DATE(created_at) = '{ds}'
AND status != 'cancelled'
GROUP BY DATE(created_at), customer_id
""")
print(f"Recomputed partition dt={ds}")
Backfill Progress Tracking
Store backfill state in a tracking table for observability:
# Create tracking table in StarRocks (one-time setup)
CREATE_TRACKING_TABLE = """
CREATE TABLE IF NOT EXISTS sales.backfill_tracking (
job_name VARCHAR(128) NOT NULL,
table_name VARCHAR(128) NOT NULL,
ds DATE NOT NULL,
status VARCHAR(32) NOT NULL, -- pending/running/done/failed
loaded_rows BIGINT,
started_at DATETIME,
finished_at DATETIME
) PRIMARY KEY(job_name, table_name, ds)
DISTRIBUTED BY HASH(ds) BUCKETS 4
PROPERTIES("enable_persistent_index" = "true");
"""
@task
def update_backfill_tracking(
ds: str,
table: str,
status: str,
loaded_rows: int = None,
):
hook = MySqlHook(mysql_conn_id="starrocks_prod")
now = datetime.utcnow().strftime("%Y-%m-%d %H:%M:%S")
hook.run(f"""
INSERT INTO sales.backfill_tracking
(job_name, table_name, ds, status, loaded_rows,
started_at, finished_at)
VALUES (
'range_backfill', '{table}', '{ds}', '{status}',
{loaded_rows or 'NULL'},
{'NULL' if status != 'running' else f"'{now}'"},
{'NULL' if status in ('running', 'pending') else f"'{now}'"}
)
""")
Query progress:
-- Check backfill progress
SELECT
ds,
status,
loaded_rows,
TIMESTAMPDIFF(SECOND, started_at, finished_at) AS duration_sec
FROM sales.backfill_tracking
WHERE job_name = 'range_backfill'
AND table_name = 'orders'
ORDER BY ds;
-- Summary
SELECT status, COUNT(*) AS days, SUM(loaded_rows) AS total_rows
FROM sales.backfill_tracking
WHERE job_name = 'range_backfill'
GROUP BY status;
Parallel Backfill with Concurrency Control
For faster backfills, parallelize across dates with a pool limit:
from airflow.providers.mysql.hooks.mysql import MySqlHook
from concurrent.futures import ThreadPoolExecutor, as_completed
import time
@task
def parallel_backfill(
pending_dates: list[str],
table: str,
db: str,
max_workers: int = 4, # max concurrent loads
):
"""Process multiple dates in parallel, up to max_workers."""
hook = MySqlHook(mysql_conn_id="starrocks_prod")
def load_one(ds: str) -> dict:
label = f"backfill_{table}_{ds.replace('-', '')}"
try:
hook.run(f"""
LOAD LABEL {db}.{label} (
DATA INFILE("s3a://datalake/{table}/dt={ds}/*.parquet")
INTO TABLE {table} FORMAT AS "parquet"
)
WITH BROKER ("aws.s3.use_instance_profile" = "true", "aws.s3.region" = "us-east-1")
PROPERTIES ("timeout" = "3600")
""")
# Poll for completion
for _ in range(120):
rows = hook.get_records(f"SHOW LOAD FROM {db} WHERE LABEL = '{label}'")
if rows and rows[0][2] == "FINISHED":
return {"ds": ds, "status": "ok"}
elif rows and rows[0][2] == "CANCELLED":
return {"ds": ds, "status": "failed", "error": rows[0][7]}
time.sleep(30)
return {"ds": ds, "status": "timeout"}
except Exception as e:
return {"ds": ds, "status": "error", "error": str(e)}
results = []
with ThreadPoolExecutor(max_workers=max_workers) as pool:
futures = {pool.submit(load_one, ds): ds for ds in pending_dates}
for fut in as_completed(futures):
result = fut.result()
results.append(result)
print(f" {result['ds']}: {result['status']}")
failed = [r for r in results if r["status"] != "ok"]
if failed:
raise RuntimeError(f"Backfill failed for {len(failed)} dates: {failed}")
return results
Anti-Patterns
catchup=True on production incremental DAGs — backfill and live ingestion run simultaneously, causing watermark conflicts; use a separate dedicated backfill DAG.- Using
INSERT INTO (append) for backfill — re-running appends duplicates; always use INSERT OVERWRITE for partition replacement.
- No
max_active_runs=1 on backfill DAG — concurrent backfill runs race on the same partition; one will overwrite the other's data.
- Backfilling without pre-checking which dates are done — re-triggering FINISHED labels is safe (idempotent), but wastes resources; skip already-FINISHED labels.
- Parallel backfill without BE resource check — many concurrent Broker Loads saturate BE disk I/O; limit to
max_workers <= BE count.
- Deleting partitions before re-loading — if the new load fails, the data is gone; use INSERT OVERWRITE which only replaces on success.
References
- Airflow catchup:
airflow.apache.org/docs/apache-airflow/stable/dag-run.html#catchup
- StarRocks INSERT OVERWRITE:
docs.starrocks.io/docs/loading/InsertInto/
- StarRocks partition management:
docs.starrocks.io/docs/table_design/Data_distribution/
- Related skills:
[[airflow-starrocks-pipeline]], [[airflow-starrocks-etl-best-practices]], [[starrocks-broker-load]], [[starrocks-partitioning]]