Large import - Nelkap

A customer uploads a file of 80,000 contacts. Importing it in one request takes too long, and if it stops at row 50,000, starting over imports the first half twice. With Nelkap, the import is split into batches, and each batch gets its own run. Batches go a few at a time, so your database never takes the whole file at once. A batch that stops goes on by itself, and the user gets an email when the last one is in.

An import of 80 batches: a grid of batch runs, most completed, a few running, the rest waiting their turn, with 52 of 80 done.

Write the workflows

The import takes two workflows: one that starts the batches, and one that imports a batch.

nelkap/workflows/import-contacts.ts

import { sendImportDoneEmail } from '../../lib/email';
import {
  importRows,
  markBatchDone,
  splitIntoBatches,
} from '../../lib/imports';
import { nelkap } from '../client';

type ContactImport = { importId: string; email: string };

export async function importContacts(
  contactImport: ContactImport,
) {
  const batchIds = await nelkap.step.run('batches', () =>
    splitIntoBatches(contactImport.importId, 1_000),
  );

  for (const [index, batchId] of batchIds.entries()) {
    await nelkap.step.run(`start batch ${index}`, () =>
      nelkap.run(
        importBatch,
        { ...contactImport, batchId },
        { tags: [contactImport.importId] },
      ),
    );
  }
}

export async function importBatch(
  batch: ContactImport & { batchId: string },
) {
  await nelkap.step.run('import', () =>
    importRows(batch.batchId),
  );

  const batchesLeft = await nelkap.step.run('progress', () =>
    markBatchDone(batch.importId, batch.batchId),
  );

  if (batchesLeft === 0) {
    await nelkap.step.run('done email', () =>
      sendImportDoneEmail(
        batch.email,
        batch.importId,
        `import_done_${batch.importId}`,
      ),
    );
  }
}

importContacts takes the import and the user’s email address:

It splits the file into batches of 1,000 rows, and saves their IDs.
It starts a run of importBatch for each batch. Each start is a step, so a run that picks up halfway doesn’t start the same batch twice. Every batch run is tagged with the import’s ID.

importBatch takes one batch:

It imports the batch’s rows.
It marks the batch as done, and counts the batches left.
If none is left, it emails the user that the import is done.

Write the functions it calls

Replace the functions from lib with your own. Three details keep the import exact:

importRows writes each contact with an upsert, keyed by its email address. A batch cut off and run again then updates the same contacts instead of adding them twice.
markBatchDone marks the batch as done in your database, then returns how many batches of the import aren’t done. Marking a batch done twice changes nothing, so the count stays right when the step runs again.
sendImportDoneEmail passes an idempotency key to your email provider. Two batches that finish at the same moment can both see no batch left, and the key makes sure the user gets one email.

The same table shows the user the import’s progress in your app: the batches done, out of the total.

Start the import

nelkap/index.ts

import { nelkap } from './client';
import {
  importBatch,
  importContacts,
} from './workflows/import-contacts';

nelkap.registerWorkflows([
  { name: 'importContacts', handler: importContacts },
  { name: 'importBatch', handler: importBatch, concurrency: 10 },
]);

export { importContacts, nelkap };

Then start a run once the file is uploaded:

await nelkap.run(
  importContacts,
  { importId: contactImport.id, email: user.email },
  { tags: [contactImport.id] },
);

See it run

Filter your project’s runs by the import’s ID: the tag lists the import’s run and every batch run. Ten batches run at a time, and the others show as pending until a place frees up. If a batch fails, its run shows the error, and the import waits for it. Fix the cause, and replay the batch’s run: it imports the batch, and sends the email if it was the last one.

Import bigger files

importContacts holds one step per batch, and a run holds up to 100 steps. With batches of 1,000 rows, that’s up to 99,000 rows. For more, make the batches bigger, or add a level: start a run per group of batches, and let each of those start its batches. See Loops and parallel steps.

Ideas to extend it

Tell the user when a batch fails: catch the StepError around the import step, and email them before you throw it again.
Record the share of valid rows in each batch with step.score, to chart the quality of your imports over time.

​Write the workflows

​Write the functions it calls

​Start the import

​See it run

​Import bigger files

​Ideas to extend it