Data Processing & ETL

Build complex data pipelines that process large datasets without timeouts.

Why Trigger.dev for ETL?

• No execution time limits — run multi-hour transformations
• Automatic retries — recover from API failures mid-pipeline
• Real-time progress — stream row-by-row status to your frontend
• Parallel processing — scale with configurable concurrency

Pattern 1: CSV Import with Real-Time Progress

import { schemaTask } from "@trigger.dev/sdk";
import { z } from "zod";

// Child task: process individual rows
export const processCSVRow = schemaTask({
  id: "process-csv-row",
  schema: z.object({
    row: z.record(z.string()),
    rowNumber: z.number(),
  }),
  run: async ({ row, rowNumber }) => {
    // Process row, save to database
    await db.insert(table).values(transformRow(row));
    metadata.parent.increment("processedRows", 1);
    return { success: true, rowNumber };
  },
});

// Parent task: coordinate
export const importCSV = schemaTask({
  id: "import-csv",
  schema: z.object({ fileUrl: z.string() }),
  run: async ({ fileUrl }) => {
    metadata.set("status", "downloading");
    const csv = await fetchCSV(fileUrl);
    const rows = parseCSV(csv);

    metadata.set("status", "processing").set("totalRows", rows.length);

    const results = await processCSVRow.batchTriggerAndWait(
      rows.map((row, i) => ({ payload: { row, rowNumber: i + 1 } }))
    );

    metadata.set("status", "complete");
    return { imported: results.filter(r => r.ok).length };
  },
});

Pattern 2: Multi-Source ETL (Coordinator)

export const etlPipeline = task({
  id: "etl-pipeline",
  run: async (payload: { date: string }) => {
    // Parallel extraction from multiple sources
    const [apiData, dbData, s3Data] = await Promise.all([
      extractFromAPI.triggerAndWait({ date: payload.date }),
      extractFromDatabase.triggerAndWait({ date: payload.date }),
      extractFromS3.triggerAndWait({ date: payload.date }),
    ]);

    // Transform and load
    const transformed = await transformData.triggerAndWait({
      api: apiData.output,
      db: dbData.output,
      s3: s3Data.output,
    });

    await loadToWarehouse.triggerAndWait({ data: transformed.output });
    return { processed: transformed.output.recordCount };
  },
});

Pattern 3: Batch Enrichment with Rate Limiting

export const enrichRecords = task({
  id: "enrich-records",
  run: async (payload: { recordIds: string[] }) => {
    const records = await db.fetchRecords(payload.recordIds);

    // Process in batches, respecting API rate limits
    const results = await enrichRecord.batchTrigger(
      records.map(record => ({
        payload: { recordId: record.id },
      })),
      {
        // SDK auto-handles rate limit backoff
      }
    );

    return { total: records.length, triggered: results.runs.length };
  },
});

export const enrichRecord = task({
  id: "enrich-record",
  retry: {
    maxAttempts: 3,
    factor: 2,
  },
  run: async ({ recordId }: { recordId: string }) => {
    const enriched = await externalAPI.enrich(recordId);
    await db.update(records).set(enriched).where(eq(records.id, recordId));
    return { enriched: true };
  },
});

Pattern 4: Parallel Web Scraping

export const scrapePages = task({
  id: "scrape-pages",
  run: async (payload: { urls: string[] }) => {
    const results = await scrapePage.batchTriggerAndWait(
      payload.urls.map(url => ({ payload: { url } }))
    );

    const scraped = results
      .filter(r => r.ok)
      .map(r => r.output);

    await db.insert(scrapedContent).values(scraped);
    return { scraped: scraped.length };
  },
});

export const scrapePage = task({
  id: "scrape-page",
  machine: "small-2x",
  run: async ({ url }: { url: string }) => {
    const browser = await puppeteer.launch();
    const page = await browser.newPage();
    await page.goto(url);
    const content = await page.evaluate(() => document.body.innerText);
    await browser.close();
    return { url, content };
  },
});

Related Examples

• Realtime CSV Importer — Full implementation with Next.js frontend
• Scrape Hacker News — Firecrawl + batch processing
• Supabase Database Operations — CRUD and storage