Orchestrating Seasonal Pipelines with Airflow

A precision-agriculture pipeline does almost nothing for six months and then does everything at once. Planting compresses a season of application data into three weeks; harvest does the same in four; the satellite archive delivers on its own schedule regardless; and a grower who has been silent since March suddenly authorises four hundred fields the week before they need prescriptions. The output of this topic is a scheduling design that survives that shape: DAGs whose cadence follows the source rather than the calendar, tasks isolated per grower, derived products triggered by arrival rather than by the clock, and retries that are safe because the ingestion manifest makes them no-ops. It sits under farm data platform engineering.

The examples use Airflow because it is the most common in this space, but every pattern here translates to Prefect, Dagster or a well-disciplined set of systemd timers. What does not translate is the discipline itself: if a task is not idempotent, no scheduler will save you.

Prerequisites

• Python 3.11+, apache-airflow 2.9.* with the standard providers, or an equivalent orchestrator
• A metadata database for the orchestrator, separate from the farm data PostGIS instance
• Ingestion code that is already idempotent — see the manifest pattern in the section overview
• Working integrations to schedule: machine data APIs and satellite imagery APIs
• Per-grower credentials in a secrets backend, not in DAG code or environment variables

1. Concept: Cadence Follows the Source

Three clocks, not one

Ingestion cadence should match how the source behaves. Satellite archives publish on revisit — polling more than daily wastes requests, polling weekly loses the one clear scene in a wet fortnight. Management-system data appears when a display next has connectivity, which during operations means several times a day and out of season means weekly. Gridded weather updates on a fixed schedule and is the only genuinely daily source.

Derived products follow a fourth rule: they should not be scheduled at all. An index raster should be computed when its inputs land, a zone map when the index series is long enough, a prescription when an agronomist asks. Scheduling derived work on the clock guarantees you either recompute unchanged inputs or wait hours for outputs that could have been ready in minutes.

Idempotency is what makes scheduling boring

A scheduler’s job is to decide when, and it is only safe to be casual about when if running twice costs nothing. With a manifest in place, the awkward questions — did the 06:00 run overlap the 12:00 one, did the retry re-download the scene, did the backfill duplicate March — all have the same answer: it does not matter. Without one, every scheduling decision becomes a correctness decision, which is how teams end up with a single daily cron nobody dares touch.

Per-grower isolation

Generate one task per grower rather than looping inside a task. A revoked token then fails exactly one task instance; retries re-process only that grower; and the failure surface in the UI matches the unit a human would act on. Airflow’s dynamic task mapping expresses this directly:

from airflow.decorators import dag, task
from datetime import datetime, timedelta

@dag(
    dag_id="ingest_machine_data",
    schedule="0 2,8,14,20 * * *",     # four times daily; see the seasonal note below
    start_date=datetime(2026, 1, 1),
    catchup=False,
    max_active_runs=1,
    default_args={"retries": 3, "retry_delay": timedelta(minutes=10),
                  "retry_exponential_backoff": True, "pool": "fmis_api"},
    tags=["ingest", "machine-data"],
)
def ingest_machine_data():

    @task
    def active_growers() -> list[str]:
        """Growers with a valid authorisation, newest connection first."""
        from platform_db import list_active_growers
        growers = list_active_growers()
        assert growers, "no active grower connections — check the token store"
        return growers

    @task(max_active_tis_per_dag=8)
    def sync_grower(grower_id: str) -> dict:
        from platform_sync import sync_one_grower
        return sync_one_grower(grower_id)     # idempotent: manifest-guarded

    @task
    def check_freshness(results: list[dict]) -> None:
        from platform_qa import assert_sources_fresh
        assert_sources_fresh(source="machine_api")

    check_freshness(sync_grower.expand(grower_id=active_growers()))

ingest_machine_data()

max_active_tis_per_dag=8 bounds concurrency at the task level, and the fmis_api pool bounds it across every DAG that touches the same provider — including backfills, which is the point.

2. Step-by-Step Implementation

Step 1 — Choose the schedule from the source, and let the season modulate it

A single cron expression cannot express “four times a day in season, weekly outside it”. Two workable approaches: run the frequent schedule year-round and let a short-circuit task exit early outside the operating window, or maintain two DAGs and enable the seasonal one by a variable. The first is simpler and costs a few seconds a day:

from airflow.decorators import task
from datetime import date

SEASON_WINDOWS = {          # (start_month_day, end_month_day) in the operating hemisphere
    "plant":   ((4, 1),  (6, 15)),
    "apply":   ((5, 1),  (8, 31)),
    "harvest": ((8, 15), (11, 30)),
}


def in_any_window(today: date) -> bool:
    md = (today.month, today.day)
    return any(start <= md <= end for start, end in SEASON_WINDOWS.values())


@task.short_circuit
def only_in_season(**context) -> bool:
    """Skip the rest of the run outside the operating windows — cheaply and visibly."""
    today = context["logical_date"].date()
    active = in_any_window(today)
    if not active:
        print(f"{today} falls outside every operating window — skipping")
    return active

Using logical_date rather than date.today() keeps the decision reproducible when a run is re-triggered, which matters when someone re-runs November’s DAG in February to debug it.

Step 2 — Trigger derivation on arrival

Airflow datasets turn “compute the index when the scene lands” into a declaration rather than a sensor loop:

from airflow import Dataset

SCENE_WINDOWS = Dataset("s3://farm-platform/published/s2_windows")

@dag(dag_id="ingest_satellite", schedule="0 5,17 * * *", start_date=datetime(2026, 1, 1),
     catchup=False, tags=["ingest", "satellite"])
def ingest_satellite():
    @task(outlets=[SCENE_WINDOWS], pool="stac_api")
    def fetch_new_scenes() -> int:
        from platform_stac import sync_new_scenes
        return sync_new_scenes()          # returns count published
    fetch_new_scenes()

@dag(dag_id="derive_indices", schedule=[SCENE_WINDOWS], start_date=datetime(2026, 1, 1),
     catchup=False, tags=["derive"])
def derive_indices():
    @task
    def compute_pending_indices() -> int:
        from platform_indices import compute_missing
        return compute_missing()          # driven by the manifest, not by the trigger payload
    compute_pending_indices()

Note that the derivation task recomputes what is missing according to the manifest, not “the thing that just arrived”. The trigger is a hint that work may exist; the manifest is the truth about which work. That indirection is what makes a missed trigger self-healing rather than a permanent gap.

Step 3 — Decide catch-up deliberately

Catch-up is right when each run owns a distinct, reproducible slice of work — a daily weather grid, a monthly aggregate. It is wrong when the run’s scope comes from mutable state such as a high-water mark, because a dozen catch-up runs then all read the same mark and race. For high-water-mark ingestion, set catchup=False and write a separate, explicitly parameterised backfill DAG:

@dag(dag_id="backfill_satellite", schedule=None, start_date=datetime(2026, 1, 1),
     catchup=False, params={"start": "2026-03-01", "end": "2026-06-30", "field_ids": []},
     tags=["backfill"])
def backfill_satellite():
    @task(pool="stac_api", priority_weight=1)     # lower than the daily DAG's default
    def backfill_window(**context) -> int:
        p = context["params"]
        from platform_stac import sync_window
        return sync_window(p["start"], p["end"], p["field_ids"] or None)
    backfill_window()

The lower priority_weight inside a shared pool is what stops a six-month backfill from starving today’s ingestion — the scheduler drains live work first and fills spare slots with history.

Step 4 — Retries that mean something

Retry transport failures and rate limits; do not retry a revoked token or a schema violation. Airflow retries everything by default, so the distinction has to live in the code: raise AirflowFailException for terminal conditions so the task fails immediately rather than burning three retries over half an hour.

from airflow.exceptions import AirflowFailException

def sync_one_grower(grower_id: str) -> dict:
    try:
        return _sync(grower_id)
    except TokenRevoked as exc:
        mark_needs_reauthorisation(grower_id)
        raise AirflowFailException(f"grower {grower_id} must re-authorise") from exc

3. Key Parameters and Tuning

4. Edge Cases and Failure Modes

The DAG that succeeds and does nothing. A token expires, the sync returns zero records, every task is green. This is why the freshness assertion runs as a task inside the DAG rather than as an external monitor — a source that has produced nothing for longer than its budget fails the run loudly.

Timezone and daylight saving in cron. A schedule of 0 2 * * * in a local timezone runs twice on one autumn night and not at all on one spring night. Schedule in UTC and convert only for display. Agronomic windows are local, but they are day-scale, so the mismatch is harmless in that direction.

Backfills that outlive their code. A backfill re-running March’s ingestion in September runs today’s code against March’s data. That is usually right for ingestion and usually wrong for derived products, where the algorithm version is part of the output’s identity. Version derived outputs by code version, as in the section overview, so a recomputation lands beside the original rather than silently replacing it.

Dynamic task mapping over an unbounded list. Mapping over four hundred growers creates four hundred task instances per run; mapping over fields creates tens of thousands and the scheduler’s metadata database becomes the bottleneck. Map over the unit a human would retry — the grower — and loop within it.

Sensors that block a worker slot. A sensor waiting six hours in the default mode occupies a worker the whole time. Use deferrable operators or reschedule mode so the slot is released between polls; during harvest those slots are the scarce resource.

Task logs that outgrow the disk. Imagery tasks log per-scene detail, and at four hundred fields a season the logs outgrow the metadata volume before the data does. Ship logs to object storage and set a retention policy on day one.

5. Verification and Output Validation

The pipeline’s own tests are the cheapest place to prove the scheduling assumptions hold.

from datetime import date

def test_season_windows_are_sane():
    assert in_any_window(date(2026, 5, 20)), "May must be in season"
    assert not in_any_window(date(2026, 1, 15)), "January must be out of season"


def test_sync_is_idempotent(tmp_platform):
    first = sync_one_grower("grower-1")
    second = sync_one_grower("grower-1")
    assert second["published"] == 0, (
        f"second sync published {second['published']} artefacts — manifest is not being consulted")
    assert first["published"] > 0, "first sync published nothing — fixture is empty"


def test_terminal_errors_do_not_retry(monkeypatch):
    monkeypatch.setattr("platform_sync._sync", _raise_token_revoked)
    with pytest.raises(AirflowFailException):
        sync_one_grower("grower-2")

The second test is the important one, and it is the test most teams do not have. Idempotency is asserted about once and then assumed forever; a refactor that moves the manifest insert outside the publishing transaction breaks it silently, and the only symptom is a slowly duplicating archive.

In production, verify the same property from the outside: once a week, count manifest rows against distinct published object keys. They must match exactly.

6. Integration with the Broader Pipeline

Orchestration is the layer that makes the rest of this section a system. It drives machine data APIs and satellite imagery APIs on their own cadences, writes through PostGIS schema design for farm data, and triggers the index and zone work that band math and raster algebra and management zone classification implement. Where an individual job is itself parallel — a multi-flight orthomosaic run, for instance — the internal concurrency belongs to the job, as in batch and async orthomosaic processing; the orchestrator’s role is to start it once and know whether it finished.

Frequently Asked Questions

Is Airflow overkill for one farm? Usually, yes. For a single operation, a cron entry calling an idempotent script and a freshness check that emails on staleness covers it. The threshold is roughly when you have more than a handful of sources, need per-grower isolation, or want a backfill you can start without SSH access.

Where should the credentials live? In a secrets backend the scheduler reads at task run time, keyed per grower. Not in Airflow Variables, which are visible to anyone with UI access, and not in the DAG file, which is in version control.

How do I test a DAG without a live provider? Split the DAG file from the work. Every task in the examples above is a thin wrapper around a function in an ordinary Python module, which is where the tests live. The DAG file itself only needs a structural test — that it imports, has no cycles, and every task has the pool and retry settings the policy requires.

This topic is part of Farm Data Platform Engineering: APIs, Storage & Orchestration — see there for the manifest and storage patterns the scheduling above depends on.

Related

• Writing Idempotent Airflow Tasks for Imagery Ingestion — staging keys, atomic promotion and the manifest insert in one transaction
• Backfilling a Season of Field Imagery Without Duplicates — a parameterised backfill that shares a pool with live ingestion
• Machine Data APIs for Management-System Integration — the polling contract these DAGs schedule
• PostGIS Schema Design for Farm Data — where the manifest that makes retries safe actually lives
• Batch and Async Orthomosaic Processing — parallelism inside a single job, as distinct from scheduling between jobs