Node/More/Examples/

ETL Data Pipeline

A diamond-shaped extract → transform → load DAG with progress reporting, error inspection, and skip-on-failure cascading.

A classic ETL job: pull rows from a source, run two independent transforms over the staged data, then load the merged result. Modeling it as a workflow gives you dependency ordering, per-step retries, and a queryable run you can monitor.

Project structure

code

pipeline/
  pipeline.ts  # the four tasks + the DAG
  monitor.ts   # analyze + inspect a run
  worker.ts    # the worker process

pipeline.ts

Each step coordinates through a staging store (the DB, S3, a temp dir) rather than passing results down the graph — the standard ETL shape. load runs only after both transforms complete. The long transform reports progress as it goes.

import { Queue, currentJob } from "@byteveda/flexiq";

export const queue = new Queue({ dbPath: "pipeline.db" });

queue.task("extract", async (runId: string) => {
  const rows = await source.read();
  await staging.write(runId, "raw", rows);
}, { maxRetries: 3 });

queue.task("clean", async (runId: string) => {
  const rows = await staging.read(runId, "raw");
  const job = currentJob();
  const out = [];
  for (let i = 0; i < rows.length; i++) {
    out.push(normalize(rows[i]));
    if (i % 1000 === 0) job?.setProgress(Math.round((i / rows.length) * 100));
  }
  await staging.write(runId, "clean", out);
});

queue.task("enrich", async (runId: string) => {
  const rows = await staging.read(runId, "raw");
  await staging.write(runId, "enrich", await enrichAll(rows));
}, { rateLimit: "200/m" }); // enrichment calls a metered API

queue.task("load", async (runId: string) => {
  const merged = merge(
    await staging.read(runId, "clean"),
    await staging.read(runId, "enrich"),
  );
  await warehouse.load(merged);
});

export function runPipeline(runId: string) {
  // `queueDefault`/`params` are submit-time options, so build the workflow
  // without the chained `.submit()` and pass the builder to
  // `queue.workflows.submit(builder, options)` instead.
  const workflow = queue.workflows
    .define("etl", 1)
    .step("extract", "extract", { args: [runId] })
    .step("clean", "clean", { args: [runId], after: "extract" })
    .step("enrich", "enrich", { args: [runId], after: "extract", queue: "io" })
    .step("load", "load", { args: [runId], after: ["clean", "enrich"], priority: 10 });
  return queue.workflows.submit(workflow, { queueDefault: "default", params: { runId } });
}

monitor.ts

analyze gives you a structural view of the run; getJobErrors surfaces the failure history of any step's underlying job.

import { queue } from "./pipeline";

export async function report(runId: string) {
  const analysis = queue.workflows.analyze(runId);
  if (!analysis) return;

  console.log("critical path:", analysis.criticalPath().join(" → "));
  console.log("stats:", analysis.stats());

  // Inspect the failure history of a specific step.
  const node = analysis.node("enrich");
  if (node?.jobId) {
    for (const err of await queue.getJobErrors(node.jobId)) {
      console.log(`enrich attempt ${err.attempt}: ${err.error}`);
    }
  }
}

If a required step fails, its still-pending dependents are skipped and the run ends completed_with_failures (or failed). Add a condition: "on_failure" step after the at-risk node to run cleanup or alerting on that path.

Running it

bash

node worker.ts   # runWorker({ queues: ["default", "io"] })

bash

node -e "import('./pipeline.ts').then(p => p.runPipeline(crypto.randomUUID()))"

Key patterns

Pattern	Where
Diamond dependency DAG	`.step(..., { after })`
Per-step queue / priority	`step` options
Progress reporting	`currentJob().setProgress`
Structural analysis	`workflows.analyze().criticalPath()`
Per-attempt error history	`queue.getJobErrors`
Skip-on-failure cascade	required step failure ⇒ pending dependents skipped

pipeline.ts

import { Queue, currentJob } from "@byteveda/flexiq";

export const queue = new Queue({ dbPath: "pipeline.db" });

queue.task("extract", async (runId: string) => {
  const rows = await source.read();
  await staging.write(runId, "raw", rows);
}, { maxRetries: 3 });

queue.task("clean", async (runId: string) => {
  const rows = await staging.read(runId, "raw");
  const job = currentJob();
  const out = [];
  for (let i = 0; i < rows.length; i++) {
    out.push(normalize(rows[i]));
    if (i % 1000 === 0) job?.setProgress(Math.round((i / rows.length) * 100));
  }
  await staging.write(runId, "clean", out);
});

queue.task("enrich", async (runId: string) => {
  const rows = await staging.read(runId, "raw");
  await staging.write(runId, "enrich", await enrichAll(rows));
}, { rateLimit: "200/m" }); // enrichment calls a metered API

queue.task("load", async (runId: string) => {
  const merged = merge(
    await staging.read(runId, "clean"),
    await staging.read(runId, "enrich"),
  );
  await warehouse.load(merged);
});

export function runPipeline(runId: string) {
  // `queueDefault`/`params` are submit-time options, so build the workflow
  // without the chained `.submit()` and pass the builder to
  // `queue.workflows.submit(builder, options)` instead.
  const workflow = queue.workflows
    .define("etl", 1)
    .step("extract", "extract", { args: [runId] })
    .step("clean", "clean", { args: [runId], after: "extract" })
    .step("enrich", "enrich", { args: [runId], after: "extract", queue: "io" })
    .step("load", "load", { args: [runId], after: ["clean", "enrich"], priority: 10 });
  return queue.workflows.submit(workflow, { queueDefault: "default", params: { runId } });
}

monitor.ts

analyze gives you a structural view of the run; getJobErrors surfaces the failure history of any step's underlying job.

import { queue } from "./pipeline";

export async function report(runId: string) {
  const analysis = queue.workflows.analyze(runId);
  if (!analysis) return;

  console.log("critical path:", analysis.criticalPath().join(" → "));
  console.log("stats:", analysis.stats());

  // Inspect the failure history of a specific step.
  const node = analysis.node("enrich");
  if (node?.jobId) {
    for (const err of await queue.getJobErrors(node.jobId)) {
      console.log(`enrich attempt ${err.attempt}: ${err.error}`);
    }
  }
}

Key patterns

Pattern	Where
Diamond dependency DAG	`.step(..., { after })`
Per-step queue / priority	`step` options
Progress reporting	`currentJob().setProgress`
Structural analysis	`workflows.analyze().criticalPath()`
Per-attempt error history	`queue.getJobErrors`
Skip-on-failure cascade	required step failure ⇒ pending dependents skipped