Distributed ETL with DuckDB-WASM (Node.js)

This tutorial builds a small but real ETL job: an insurance company’s quarterly claims, split by region, distributed across the DCP network, and summarized into a company-wide loss report, using DuckDB compiled to WebAssembly as the query engine inside each worker.

The insurance example is deliberately small; the capability behind it isn’t. DuckDB-WASM gives every DCP worker a real, embedded SQL engine, able to query Parquet and CSV directly and run joins, aggregations, and window functions against them. That’s the same class of workload most ETL pipelines run today on a managed Spark cluster (Databricks and similar): SQL-driven transformation over large tabular data. Doing it here instead means that SQL logic and its distribution across machines both run on DCP, at a fraction of the cost of a managed cluster you have to provision, size, and pay for whether or not it’s busy.

It assumes you’ve already seen the basic DCP job tutorial; this one reuses the same shape (input set → work function → compute.for → events → results) and adds one new ingredient: real SQL queries running inside a sandboxed worker.

Prerequisites

• Node.js (LTS)

• npm i dcp-client

• Keystore files in ~/.dcp (see the basic tutorial if you don’t have these yet)

• A DuckDB-WASM build, published to the DCP package manager as duckdbwasm and available to every job: just add job.requires(['duckdbwasm/duckdb-wasm.js']) as shown below, nothing to set up yourself. It’s WASM rather than the native duckdb npm package because DCP workers run in a sandboxed, browser-like environment and can’t load native binaries. The published build already includes the extensions needed for common ETL workloads; if yours needs one that isn’t there, email dan@dcp.dev and it’ll get added.

The scenario

A regional insurer wants a quarterly loss summary (claim count, total paid, average claim size) broken down by policy type, for every region, then rolled up into one company-wide number, with any individually large claim flagged for review. Each region’s claims are independent of every other region’s, which makes this an ideal shape for distributed compute: one region’s claims are one unit of work, and a DCP worker can summarize a region completely on its own using nothing but the rows it was handed and the review threshold every worker shares.

Building it, concept by concept

Input set

One element per region. Each element carries that region’s raw claims, nothing shared, nothing another region’s worker needs to know about:

const inputSet = [
  {
    region: 'northeast',
    claims: [
      { policyType: 'auto', amount: 4250.00,  status: 'paid'   },
      { policyType: 'auto', amount: 900.00,   status: 'denied' },
      { policyType: 'home', amount: 12800.50, status: 'paid'   },
      { policyType: 'auto', amount: 3100.75,  status: 'paid'   },
      { policyType: 'home', amount: 2200.00,  status: 'paid'   },
    ],
  },
  {
    region: 'southeast',
    claims: [
      { policyType: 'auto',  amount: 5100.00,  status: 'paid'   },
      { policyType: 'home',  amount: 800.00,   status: 'denied' },
      { policyType: 'auto',  amount: 2750.25,  status: 'paid'   },
      { policyType: 'flood', amount: 41200.00, status: 'paid'   },
    ],
  },
  {
    region: 'midwest',
    claims: [
      { policyType: 'auto', amount: 3300.00, status: 'paid'   },
      { policyType: 'home', amount: 9100.00, status: 'paid'   },
      { policyType: 'auto', amount: 600.00,  status: 'denied' },
      { policyType: 'home', amount: 1450.00, status: 'paid'   },
      { policyType: 'auto', amount: 2900.00, status: 'paid'   },
    ],
  },
];

A real job would pull each region’s claims from wherever your claims system exports them (Parquet or CSV extracts are the common case; duckdb-wasm can query those directly out of an in-memory buffer via db.registerFileBuffer(), no filesystem needed). Small inline arrays keep this tutorial’s data on the page; the query logic below doesn’t change if each element becomes “a buffer of Parquet bytes” instead of “an array of objects.”

Static arguments

Not everything a work function needs varies by region. The threshold above which a claim counts as “large” and gets flagged for review is a single business rule the whole company shares, not something to repeat inside every inputSet element or hardcode inside workFunction itself. compute.for takes a third argument for exactly this: an array of values passed once, delivered to every invocation of the work function alongside that invocation’s own input-set element.

const LARGE_CLAIM_THRESHOLD = 10000;

workFunction picks it up as an extra parameter after the input-set element, and job creation passes it through as a one-element array:

async function workFunction({ region, claims }, largeClaimThreshold) { /* ... */ }

const job = compute.for(inputSet, workFunction, [LARGE_CLAIM_THRESHOLD]);

Every worker gets the same largeClaimThreshold, without it ever appearing in the sharded data. Static arguments are the right home for anything like this: a config value, a lookup table, a shared constant, whatever every slice needs identically rather than individually.

Work function

This is where the actual ETL work happens, once per region, inside a sandboxed DCP worker. It loads that region’s claims into an in-memory DuckDB table, then runs a real SQL query against it, the same query you’d hand to a data warehouse: paid claims, grouped by policy type, with count/total/average, plus how many of those claims clear the shared large-claim threshold.

async function workFunction({ region, claims }, largeClaimThreshold) {
  progress();

  const duckdb = require('duckdb-wasm.js'); // the published package, via job.requires() below
  const db = await duckdb.createDB();
  const con = db.connect();

  con.query(`
    CREATE TABLE claims (policy_type VARCHAR, amount DOUBLE, status VARCHAR)
  `);

  // Prepared statement, not string-interpolated SQL -- worth doing even
  // here, since it's the habit that matters once this is real claims data.
  const insert = con.prepare('INSERT INTO claims VALUES (?, ?, ?)');
  for (const c of claims) insert.query(c.policyType, c.amount, c.status);
  insert.close();

  // largeClaimThreshold came in as a static argument (see above), not part
  // of claims -- every region's worker applies the same threshold here.
  const stmt = con.prepare(`
    SELECT policy_type,
           COUNT(*)::INTEGER     AS claim_count,
           SUM(amount)           AS total_paid,
           ROUND(AVG(amount), 2) AS avg_claim,
           SUM(CASE WHEN amount >= ? THEN 1 ELSE 0 END)::INTEGER AS large_claim_count
    FROM claims
    WHERE status = 'paid'
    GROUP BY policy_type
    ORDER BY policy_type
  `);
  const result = stmt.query(largeClaimThreshold);
  stmt.close();

  // duckdb-wasm returns query results as Arrow rows -- schema.fields + a
  // plain toArray() turns them into ordinary JS objects.
  const fields = result.schema.fields.map((f) => f.name);
  const byPolicyType = result.toArray().map((row) => {
    const out = {};
    for (const f of fields) out[f] = row[f];
    return out;
  });

  con.close();
  return { region, byPolicyType };
}

Job creation

const job = compute.for(inputSet, workFunction, [LARGE_CLAIM_THRESHOLD]);

Job configuration

job.requires() is what makes require('duckdb-wasm.js') resolve inside the worker; without it, that line throws:

job.requires(['duckdbwasm/duckdb-wasm.js']);

job.public = {
  name: 'insuranceLossSummary',
  description: 'Per-region claims loss summary computed with DuckDB-wasm',
  link: 'https://distributive.network',
};

job.computeGroups = [{ joinKey: 'public' }];

Events

job.on('readystatechange', (ev) => console.log(`Ready state: ${ev}`));
job.on('accepted', () => console.log(`  Job id: ${job.id}\n  Awaiting results...`));
job.on('result', (ev) => console.log(`  ${ev.result.region} done`));
job.on('error', (error) => console.error('  Job error:', error));
job.on('nofunds', (ev) => console.log(ev));

Execution

const results = await job.exec();

Result post-processing

Each region came back as its own little report. Rolling them into one company-wide number is the same aggregation idea as the SQL above, just done in plain JS across the (small) number of regions rather than across the (large) number of claims:

const companyWide = {};
for (const { region, byPolicyType } of results) {
  console.log(`\n${region}:`);
  for (const row of byPolicyType) {
    console.log(`  ${row.policy_type.padEnd(6)} claims=${row.claim_count}  paid=$${row.total_paid.toFixed(2)}  avg=$${row.avg_claim}  large=${row.large_claim_count}`);
    const agg = companyWide[row.policy_type] || { claimCount: 0, totalPaid: 0, largeClaimCount: 0 };
    agg.claimCount += row.claim_count;
    agg.totalPaid += row.total_paid;
    agg.largeClaimCount += row.large_claim_count;
    companyWide[row.policy_type] = agg;
  }
}

console.log('\nCompany-wide, all regions:');
for (const [policyType, agg] of Object.entries(companyWide)) {
  console.log(`  ${policyType.padEnd(6)} claims=${agg.claimCount}  paid=$${agg.totalPaid.toFixed(2)}  large=${agg.largeClaimCount}`);
}

Full listing

async function main() {
  const compute = require('dcp/compute');

  /* INPUT SET */
  const inputSet = [
    {
      region: 'northeast',
      claims: [
        { policyType: 'auto', amount: 4250.00,  status: 'paid'   },
        { policyType: 'auto', amount: 900.00,   status: 'denied' },
        { policyType: 'home', amount: 12800.50, status: 'paid'   },
        { policyType: 'auto', amount: 3100.75,  status: 'paid'   },
        { policyType: 'home', amount: 2200.00,  status: 'paid'   },
      ],
    },
    {
      region: 'southeast',
      claims: [
        { policyType: 'auto',  amount: 5100.00,  status: 'paid'   },
        { policyType: 'home',  amount: 800.00,   status: 'denied' },
        { policyType: 'auto',  amount: 2750.25,  status: 'paid'   },
        { policyType: 'flood', amount: 41200.00, status: 'paid'   },
      ],
    },
    {
      region: 'midwest',
      claims: [
        { policyType: 'auto', amount: 3300.00, status: 'paid'   },
        { policyType: 'home', amount: 9100.00, status: 'paid'   },
        { policyType: 'auto', amount: 600.00,  status: 'denied' },
        { policyType: 'home', amount: 1450.00, status: 'paid'   },
        { policyType: 'auto', amount: 2900.00, status: 'paid'   },
      ],
    },
  ];

  /* STATIC ARGUMENTS -- shared by every worker, not part of the input set */
  const LARGE_CLAIM_THRESHOLD = 10000;

  /* WORK FUNCTION */
  async function workFunction({ region, claims }, largeClaimThreshold) {
    progress();

    const duckdb = require('duckdb-wasm.js');
    const db = await duckdb.createDB();
    const con = db.connect();

    con.query(`
      CREATE TABLE claims (policy_type VARCHAR, amount DOUBLE, status VARCHAR)
    `);
    const insert = con.prepare('INSERT INTO claims VALUES (?, ?, ?)');
    for (const c of claims) insert.query(c.policyType, c.amount, c.status);
    insert.close();

    const stmt = con.prepare(`
      SELECT policy_type,
             COUNT(*)::INTEGER     AS claim_count,
             SUM(amount)           AS total_paid,
             ROUND(AVG(amount), 2) AS avg_claim,
             SUM(CASE WHEN amount >= ? THEN 1 ELSE 0 END)::INTEGER AS large_claim_count
      FROM claims
      WHERE status = 'paid'
      GROUP BY policy_type
      ORDER BY policy_type
    `);
    const result = stmt.query(largeClaimThreshold);
    stmt.close();

    const fields = result.schema.fields.map((f) => f.name);
    const byPolicyType = result.toArray().map((row) => {
      const out = {};
      for (const f of fields) out[f] = row[f];
      return out;
    });

    con.close();
    return { region, byPolicyType };
  }

  /* JOB CREATION */
  const job = compute.for(inputSet, workFunction, [LARGE_CLAIM_THRESHOLD]);

  /* JOB CONFIGURATION */
  job.requires(['duckdbwasm/duckdb-wasm.js']);
  job.public = {
    name: 'insuranceLossSummary',
    description: 'Per-region claims loss summary computed with DuckDB-wasm',
    link: 'https://distributive.network',
  };
  job.computeGroups = [{ joinKey: 'public' }];

  /* EVENTS */
  job.on('readystatechange', (ev) => console.log(`Ready state: ${ev}`));
  job.on('accepted', () => console.log(`  Job id: ${job.id}\n  Awaiting results...`));
  job.on('result', (ev) => console.log(`  ${ev.result.region} done`));
  job.on('error', (error) => console.error('  Job error:', error));
  job.on('nofunds', (ev) => console.log(ev));

  /* EXECUTION */
  const results = await job.exec();

  /* RESULT POST-PROCESSING */
  const companyWide = {};
  for (const { region, byPolicyType } of results) {
    console.log(`\n${region}:`);
    for (const row of byPolicyType) {
      console.log(`  ${row.policy_type.padEnd(6)} claims=${row.claim_count}  paid=$${row.total_paid.toFixed(2)}  avg=$${row.avg_claim}  large=${row.large_claim_count}`);
      const agg = companyWide[row.policy_type] || { claimCount: 0, totalPaid: 0, largeClaimCount: 0 };
      agg.claimCount += row.claim_count;
      agg.totalPaid += row.total_paid;
      agg.largeClaimCount += row.large_claim_count;
      companyWide[row.policy_type] = agg;
    }
  }

  console.log('\nCompany-wide, all regions:');
  for (const [policyType, agg] of Object.entries(companyWide)) {
    console.log(`  ${policyType.padEnd(6)} claims=${agg.claimCount}  paid=$${agg.totalPaid.toFixed(2)}  large=${agg.largeClaimCount}`);
  }
}
require('dcp-client').init('https://scheduler.distributed.computer').then(main).catch(console.error);

Running it

node claims-loss-summary.js

Expected output (dollar figures will match the sample data above):

Ready state: ...
  Job id: ...
  Awaiting results...
  northeast done
  southeast done
  midwest done

northeast:
  auto   claims=2  paid=$7350.75  avg=$3675.38  large=0
  home   claims=2  paid=$15000.50  avg=$7500.25  large=1

southeast:
  auto   claims=2  paid=$7850.25  avg=$3925.13  large=0
  flood  claims=1  paid=$41200.00  avg=$41200  large=1

midwest:
  auto   claims=2  paid=$6200.00  avg=$3100  large=0
  home   claims=2  paid=$10550.00  avg=$5275  large=0

Company-wide, all regions:
  auto   claims=6  paid=$21401.00  large=0
  home   claims=4  paid=$25550.50  large=1
  flood  claims=1  paid=$41200.00  large=1

Going further

• Bigger, real data: swap the inline claims arrays for Parquet or CSV files exported from your claims system, and load them with db.registerFileBuffer(name, bytes) + read_parquet('name') / read_csv_auto('name') inside the work function instead of INSERT statements. The rest of the query doesn’t change.

• Wider SQL: this tutorial’s query deliberately stays inside DOUBLE and an explicit ::INTEGER cast on COUNT(*), so every result column comes back as a plain JS number. Reach for DATE/TIMESTAMP columns or SUM over large INTEGER columns and you’ll get values back wrapped as Arrow types (epoch milliseconds for dates, a DecimalBigNum-style object for wide sums) that need one extra conversion step, keyed off result.schema.fields[i].type.typeId. Worth knowing about before it surprises you, not a reason to avoid richer SQL.

• Sharding by something other than region: the input set is just an array: shard by policy type, by month, by claims adjuster, whatever divides your real workload into independent units. One slice, one worker, one self-contained answer is the whole pattern.

• More static arguments: the same pattern from LARGE_CLAIM_THRESHOLD extends to anything every worker needs identically, a currency-conversion table, a set of valid policy types to validate against, per-tenant config. Pass it once as the third argument to compute.for, not duplicated into every input-set element.