How to migrate a Google Cloud Composer DAG

To migrate a Google Cloud Composer DAG, collect the DAG source and its runtime dependencies, then map the data interval, dependencies, and delivery behavior for one feed. Keep operators that already call reliable external jobs where they still fit. Move the feed's coordination to a small Python runner or another Airflow environment, and prove that reruns cover the same data interval.

How do you retry a Cloud Composer DAG when the report partition is missing?

The Composer DAG shows success, but a report partition is missing. Inspect required task outcomes, the stored data interval and the external job result before treating a successful cleanup leaf as evidence of delivery.

Recovery. Reuse the original run's stored interval start and end on retry. Publish only after that interval's partition passes validation and required tasks permit it. On an ambiguous BigQuery submission, look up the saved project, location and job ID before retrying a write.

Cutover check. Compare actual legacy interval bounds, rendered parameters, partition keys and delivery receipts against a shadow backfill and scheduled run. Verify that a retry covers the same interval and the observed local timetable reproduces the daylight-saving boundary.

Replace an Airflow DAG with an explicit run contract

This synthetic Composer DAG has four tasks: wait for the daily input, submit a BigQuery transform, validate its partition, and publish a ready receipt. A cleanup task runs independently after any terminal outcome. Use the old run's rendered task fields and actual external job receipt to reconstruct the workflow. Airflow's run status is derived from leaf tasks, so a successful all_done cleanup leaf can coexist with a failed intermediate task. Confirm the downstream partition rather than using DAG success as your only evidence. Airflow DAG runs and leaf status

Execute the same clock interval after a retry

For the example's observed Los Angeles midnight timetable, this Python check reproduces the UTC bounds of the March transition. It uses the timezone database available to Python 3.9 or later. Use the actual timetable's observed bounds for manual, asset-triggered or custom runs rather than forcing them through this daily helper.

from datetime import date, datetime, time, timedelta, timezone
from zoneinfo import ZoneInfo

def bounds(day):
    zone = ZoneInfo("America/Los_Angeles")
    start = datetime.combine(day, time.min, zone)
    end = datetime.combine(day + timedelta(days=1), time.min, zone)
    return start.astimezone(timezone.utc), end.astimezone(timezone.utc)

start, end = bounds(date(2026, 3, 8))
assert start.isoformat() == "2026-03-08T08:00:00+00:00"
assert end.isoformat() == "2026-03-09T07:00:00+00:00"
assert end - start == timedelta(hours=23)

Write the normalized interval into the run contract once, then pass it to the task that renders source path and query parameters. A run at 07:06Z and a rerun at 12:00Z both read the same admitted input and keep the same destination identity. Add one event exactly at the interval end. It belongs to the following interval under the example's half-open predicate. Include a manual run with explicit business-date parameters and an unavailable input. Check that an optional skipped branch can proceed only if the observed trigger rule allowed it.

Check external execution before switching the schedule

Stage against a separate BigQuery dataset or shadow partition. When a submit response is lost, look up the known job ID in the correct project and location and verify the recorded job configuration before interpreting its status. Use a separate job attempt for a confirmed failed run if the service requires it. Keep the per-interval publication identity stable. BigQuery managing jobs

Pause the selected legacy DAG schedule and track the last interval with a consumer-visible receipt. Record queued, running and submitted-but-not-confirmed jobs separately. Run the replacement for the first interval you actually own, then explicitly register any historical backfill. Rollback reassigns uncommitted intervals after reconciling external job results and existing receipts.

Implementation runbook

Capture the DAG, runtime and one manual plus scheduled run

Export the source for the specific DAG from the environment bucket, preferably matching the commit deployed at the observed run. The command can overwrite existing paths. Review imports, then repeat for each required deployed helper. Record requirements/provider versions and named connections from the deployed environment or manifest. No secret values are required for a parity fixture.

mkdir -p ./composer-audit
gcloud composer environments describe "$ENV" --location="$REGION" \
  --format=json > composer-environment.json
gcloud composer environments storage dags export --environment="$ENV" \
  --location="$REGION" --source="$DAG_FILE" --destination=./composer-audit

# For a deployed Airflow 2 environment:
gcloud composer environments run "$ENV" --location="$REGION" dags list-runs \
  -- -d "$DAG_ID" -o json

For Airflow 3, use the Airflow web UI or the REST API v2 endpoint GET /api/v2/dags/{dag_id}/dagRuns to list runs. Inspect task outcomes, actual data intervals and source paths in the UI or API for the installed Airflow version. Follow helpers/plugins, SQL and templated paths. Capture image and Airflow/provider versions, named Variables/Connections and secret backend references, service accounts, network routes, worker requirements, cloud jobs launched by operators, and run-level conf. Keep definitions and run metadata private. Composer export · Remote Airflow command · Airflow 2 CLI · Composer Airflow REST API access

Reproduce Airflow semantics before simplifying the scheduler

Read the timetable, schedule type and timezone, catchup setting, start date, pools, max active runs, depends_on_past and retry delays. For interval timetables, Airflow typically schedules a run after its data interval ends. Its logical date can be the interval start. Do not infer interval boundaries from actual start, ds, the date a retry ran, or an assumed version default. A trigger timetable, event/asset trigger, custom timetable and manual run can differ. Capture the actual values, rendered Jinja templates and dag_run.conf precedence for the selected flow. DAG runs and data intervals · Timetables

For each task, identify dependency outcome (all_success, failure, skips or another rule), dynamic mapping inputs, sensor poke/reschedule/deferral and timeout, task retry, wait-for-completion behavior of external operators, and XCom schema/size. A successful operator that only submits a BigQuery/Dataflow job may precede the external job completion. Capture a real job ID and receipt when downstream behavior depends on it. Inspect required leaf tasks and sink result to confirm DAG status. Airflow DAG run status

Implement one reproducible scheduled feed

1. Change the launcher to accept interval_start_utc, interval_end_utc, feed, run_mode, a transform revision and input manifest ID. For normal runs derive those from the agreed business timetable. For a backfill pass explicit reviewed intervals. Freeze the source version or BigQuery snapshot for a given manifest. A clock retry must not shift the interval.
2. Replace the file sensor with check_input(interval, deadline) returning waiting, ready, skipped or failed according to the original task and trigger rule. Pass a validated artifact path or job receipt between functions instead of local worker files or large implicit XComs. Store external job IDs so retry can check status before submitting another job.
3. Transform and stage by full interval. Use half-open time predicates if confirmed. Write a manifest with data interval, input identities, expected rows, output and reject counts, schema and code version. Make the approved publish atomic/idempotent per interval. Recover from target receipt if the scheduler dies after commit. Retry only approved transient states with the same arguments and preserve downstream run gating.
4. Compare task outcomes, actual rendered query, accepted and rejected keys, data interval, output format and receipt. Shadow a historical backfill and one normally scheduled interval. Stop the original schedule for the named feed, let already-running writes settle, and reconcile the last old interval. A rollback should explicitly target missing intervals to avoid duplicate catchup.

Test boundary, DST, replay, and skip branches

For UTC daily data, 2026-01-07T00:00Z <= event_time < 2026-01-08T00:00Z includes event E1 at Jan 7 06:00Z and excludes E2 at Jan 8 00:00Z; E2 belongs to the next run. For a Los Angeles midnight-to-midnight feed, March 8, 2026 runs from 08:00Z March 8 through 07:00Z March 9 (23 hours), and November 1 runs from 07:00Z November 1 through 08:00Z November 2 (25 hours). A fixed “minus 24 hours” should fail that fixture. Run a second time with the same interval and input version to prove one final result. For a missing upstream file test the original sensor and trigger rule, including whether a join of skipped/success tasks can publish. Crash after an external job submits and confirm the retry checks its saved receipt instead of submitting another write.

Prompt for a migration assistant

Read the sanitized DAG, helper imports, rendered task instance parameters and three traces (scheduled, manual and retried). Produce a matrix of interval, trigger rule, timeout, execution dependency, rendered input/output and external completion receipt. Build stage-only code that takes an explicit interval and tests midnight, DST, missing input, a skipped branch and restart after external job submission. Cite the DAG lines and logged values for each chosen behavior.

Try an interval boundary fixture

This Python 3 fixture models a feed whose observed data interval is half-open. Each ISO timestamp is UTC. The event at midnight belongs to the next run.

from datetime import datetime

start = datetime.fromisoformat("2026-01-07T00:00:00+00:00")
end = datetime.fromisoformat("2026-01-08T00:00:00+00:00")
events = [
    ("before", "2026-01-06T23:59:59+00:00"),
    ("inside", "2026-01-07T06:00:00+00:00"),
    ("next", "2026-01-08T00:00:00+00:00"),
]
assert [key for key, ts in events if start <= datetime.fromisoformat(ts) < end] == ["inside"]

Run two adjacent intervals against your shadow destination. Expect no overlap and no gap at midnight. If your DAG schedules in a local timezone, also check an interval across a daylight saving change.

Copy these prompts into your AI assistant

Read this sanitized DAG, its imports, and two run histories. Map each task to the system it calls and list its readiness, skip, retry, and failure rules. Explain the actual data interval, with file references. Flag rules that depend on configuration outside the DAG.

Draft a runnable replacement for this one customer feed. Take start and end as explicit arguments and write only to staging. Include tests for a missing upstream file, adjacent intervals, a backfill that is run twice, and a failed task that must block publication.

Schedule with a backfill plan

Allow one writer per interval, cap retries for transient failures, and alert on missing inputs or failed validation. A scheduled AI task can summarize fresh and missing partitions from the run manifest and assign unresolved cases to the DAG owner.

Compare a historical backfill and at least one normal run in isolation. Pause the Composer DAG after the last approved interval, let active writes finish, and start the replacement at the next interval. For rollback, inspect published intervals before unpausing the old DAG so catchup does not publish the same output twice.

References

• Google Cloud DAG management
• Airflow DAG runs, catchup, and backfill

Need help with the migration?

Explore Google Cloud Composer migration services or browse all DIY migration guides.