After the backfill of Backfills, the landing zone holds one orders.jsonl per day. booknest_mapped summarizes however many there are, a number unknown when the file is parsed:
@task
def list_files() -> list[str]: # decided at run time, not parse time
return sorted(str(p) for p in LANDING.glob("date=*/orders.jsonl"))
@task(max_active_tis_per_dagrun=2) # at most two files at once
def summarize(path: str, skip_status: str) -> dict:
with open(path, encoding="utf-8") as f:
orders = [o for o in map(json.loads, f) if o["status"] != skip_status]
return {"day": Path(path).parent.name.removeprefix("date="), "orders": len(orders),
"revenue": round(sum(o["total"] for o in orders), 2)}
...
report(summarize.partial(skip_status="cancelled").expand(path=list_files()))Output
summarize | 1 | success | 23:49:01 | 23:49:02 summarize | 0 | success | 23:49:01 | 23:49:02 summarize | 2 | success | 23:49:03 | 23:49:03 2026-06-28 179 orders 6549.09 2026-06-29 182 orders 6247.71 2026-06-30 182 orders 6622.67
Three paths made three summarize instances (map_index 0-2), and the cap of two made the third wait. report printed one line per instance in index order and returned 543 non-cancelled orders. A failed instance retries and can be cleared on its own, so keep each one idempotent and independent of its siblings.