Batched Hourly Cleanup Pipeline

Create a daily pipeline that dynamically discovers hourly log slices from a manifest and cleans them in batched runs before merging outputs.

Capabilities

Manifest-driven discovery

• Given a manifest JSON file containing a date and an hours array (for example ["00","02","05"]), the pipeline creates cleaning tasks only for listed hours and produces one cleaned file per hour under <output_dir>/<date>/clean/HH.txt; no files are created for hours not listed. @test

Batched hourly cleaning

• Cleaning tasks for a single date share a batch-friendly parameter so that the scheduler can group up to max_batch_size hourly runs into one execution; with 3 hours and max_batch_size=2, the cleaning step should run in two grouped executions while still producing three cleaned files. @test

Final aggregation

• After all hourly outputs exist, a final task concatenates them in hour order into <output_dir>/<date>/daily.txt only after dynamic discovery completes. @test

Idempotent resume

• If a cleaned hourly file already exists before the pipeline starts, it is treated as complete and not rewritten; the final aggregate still includes it. @test

Implementation

@generates

API

def build_and_run(date_str: str, manifest_path: str, output_dir: str, max_batch_size: int = 2) -> bool:
    """Runs the daily cleaning pipeline. Returns True when aggregation finishes successfully."""

Dependencies { .dependencies }

luigi { .dependency }

Provides workflow orchestration, dynamic dependencies, and batched execution.