ETL Pipeline Runner

Project roadmap

1. 01

   Design the runner

   ~2h

   Write DESIGN.md: a pipeline is defined in a config file (YAML or JSON) listing stages and their settings; each stage is a Perl module with a run method that takes the previous stage's output file and writes its own. Runs and stage states (pending, running, succeeded, failed, with attempts and errors) are stored in a database through DBI. Checkpoints are files in runs/<run id>/. Done when: DESIGN.md covers the config, the states and what resuming does.

2. 02

   Run stages and record state

   ~2.5h

   Implement the runner: create a run, execute the stages in order, and record each stage's start, end, status and error in the database. Catch errors with eval (or Try::Tiny) so a failing stage is recorded as failed and the run stops cleanly. Log every step with timestamps. Done when: a sample pipeline runs, and the database shows every stage's status.

3. 03

   Extract from an HTTP API

   ~2h

   Write an extract stage that downloads records from a JSON API with LWP::UserAgent, following pagination, with timeouts. Retry network errors, 429 and 5xx responses with increasing waits, but fail at once on 4xx errors. Write the records as JSON Lines into the stage's checkpoint file, writing to a temporary file and renaming it at the end. Done when: the extract stage downloads every page and survives a few simulated failures.

4. 04

   Transform and load

   ~2.5h

   Write a transform stage that validates and reshapes records (types, required fields, dates), sending bad records to a rejects file with the reason. Write a load stage that inserts records into a target table in batches of 500 with prepared statements, each batch in a transaction, using the source's record id as a unique key and upserting, so loading twice gives the same result. Done when: the pipeline loads the data, and running the load stage twice doesn't duplicate rows.

5. 05

   Resume failed runs

   ~2h

   Add a resume RUN_ID command that skips succeeded stages whose checkpoint files exist, and restarts from the first failed stage using the previous checkpoint. Add a lock so two copies of the same pipeline can't run at once, and a --from STAGE option to force re-running from a stage. Done when: a run failing at transform resumes at transform after a fix, without downloading again.

6. 06

   Add reporting

   ~1.5h

   Add a status command listing recent runs with their duration and stage statuses, and a show RUN_ID command with the details, record counts per stage, rejects and errors. Exit with a non-zero code when a run fails, so it can be used from cron. Done when: the status command gives a clear picture of past runs.

7. 07

   Test failure scenarios

   ~2h

   Write tests with Test::More using SQLite and a fake API (a local HTTP::Daemon or mocked user agent): a full run, a failure in each stage followed by resume, retries, idempotent load, rejects, and the lock preventing parallel runs. Done when: prove -l t/ passes.

Resources

• DocsDBI on MetaCPAN
• DocsLWP::UserAgent on MetaCPAN
• DocsTry::Tiny on MetaCPAN
• DocsTest::More on MetaCPAN
• ArticleExponential backoff and jitter (AWS)