CSV import in chunks

Goal: accept a CSV upload of any size, answer at once, and import it in chunks of 500 rows. Several workers can import chunks in parallel, and a failed chunk is retried on its own.

Runnable project

Every file below is imported from examples/csv-import, which has a Makefile and an e2e.sh that runs the whole flow. Clone the repository, cd examples/csv-import, and run make install up.

Key files

• src/ImportChunkHandler.php
• bin/worker.php
• public/index.php
• docker-compose.yml

View source on GitHub

Files

Redis and src/Hermes.php as in the welcome email tutorial. Create a storage/ folder.

{
    "require": {
        "php": ">=8.4",
        "ext-pdo_sqlite": "*",
        "tomaj/hermes": "dev-master",
        "predis/predis": "^2.3 || ^3.0"
    },
    "autoload": {
        "psr-4": {
            "App\\": "src/"
        }
    },
    "minimum-stability": "dev",
    "prefer-stable": true
}

<?php

declare(strict_types=1);

use App\Hermes;
use Tomaj\Hermes\Message;

require __DIR__ . '/../vendor/autoload.php';

const CHUNK = 500;

if (($_FILES['csv']['error'] ?? UPLOAD_ERR_NO_FILE) !== UPLOAD_ERR_OK) {
    http_response_code(422);
    echo "Upload a file as the csv field.\n";
    return;
}

// Keep the upload somewhere the worker can read it.
$path = __DIR__ . '/../storage/' . bin2hex(random_bytes(8)) . '.csv';
if (!is_dir(dirname($path))) {
    mkdir(dirname($path), 0o775, true);
}
move_uploaded_file($_FILES['csv']['tmp_name'], $path);

// Count data rows (the first line is the header), then queue one message per chunk.
$rows = -1;
foreach (new SplFileObject($path) as $line) {
    if ($line !== '' && $line !== false) {
        $rows++;
    }
}

$emitter = Hermes::emitter();
for ($offset = 0; $offset < $rows; $offset += CHUNK) {
    $emitter->emit(new Message('csv.import_chunk', ['path' => $path, 'offset' => $offset, 'limit' => CHUNK]));
}

http_response_code(202);
printf("%d rows queued in %d chunks.\n", $rows, (int) ceil($rows / CHUNK));

<?php

declare(strict_types=1);

namespace App;

use LimitIterator;
use PDO;
use SplFileObject;
use Tomaj\Hermes\Handler\HandlerInterface;
use Tomaj\Hermes\Handler\RetryTrait;
use Tomaj\Hermes\MessageInterface;

final class ImportChunkHandler implements HandlerInterface
{
    use RetryTrait;

    public function __construct(private PDO $db) {}

    #[\Override]
    public function handle(MessageInterface $message): bool
    {
        ['path' => $path, 'offset' => $offset, 'limit' => $limit] = $message->getPayload();

        $file = new SplFileObject($path);
        $file->setFlags(SplFileObject::READ_CSV | SplFileObject::SKIP_EMPTY | SplFileObject::READ_AHEAD);

        // Line 0 is the header, so data row N is line N + 1.
        $insert = $this->db->prepare(
            'INSERT INTO customers (email, name) VALUES (:email, :name)
             ON CONFLICT (email) DO UPDATE SET name = excluded.name',
        );

        // One transaction per chunk: a failed chunk is retried as a whole, and the upsert makes that safe.
        $this->db->beginTransaction();
        try {
            foreach (new LimitIterator($file, $offset + 1, $limit) as $row) {
                [$email, $name] = $row + [null, null];
                if ($email === null || $email === '') {
                    continue;
                }
                $insert->execute(['email' => strtolower(trim($email)), 'name' => trim((string) $name)]);
            }
            $this->db->commit();
        } catch (\Throwable $e) {
            $this->db->rollBack();
            throw $e;
        }

        return true;
    }
}

<?php

declare(strict_types=1);

use App\Hermes;
use App\ImportChunkHandler;
use Tomaj\Hermes\Dispatcher;
use Tomaj\Hermes\Shutdown\PredisShutdown;

require __DIR__ . '/../vendor/autoload.php';

if (!is_dir(__DIR__ . '/../storage')) {
    mkdir(__DIR__ . '/../storage', 0o775, true);
}

$db = new PDO('sqlite:' . __DIR__ . '/../storage/app.sqlite');
$db->setAttribute(PDO::ATTR_ERRMODE, PDO::ERRMODE_EXCEPTION);
$db->exec('CREATE TABLE IF NOT EXISTS customers (email TEXT PRIMARY KEY, name TEXT NOT NULL)');

$redis = Hermes::redis();
$driver = Hermes::driver($redis);
$driver->setShutdown(new PredisShutdown($redis));

$dispatcher = new Dispatcher($driver);
$dispatcher->registerHandler('csv.import_chunk', new ImportChunkHandler($db));
$dispatcher->handle();

Run it

composer install
mkdir -p storage
docker compose up -d redis
php bin/worker.php                    # terminal 1
php -S localhost:8002 -t public       # terminal 2

Make a test file with 1234 rows and upload it:

php -r '$f = fopen("customers.csv", "w"); fputcsv($f, ["email", "name"]); for ($i = 0; $i < 1234; $i++) { fputcsv($f, ["user$i@example.com", "User $i"]); }'
curl -F csv=@customers.csv http://localhost:8002/

1234 rows queued in 3 chunks.

Check the result:

php -r '$db = new PDO("sqlite:storage/app.sqlite"); echo $db->query("SELECT COUNT(*) FROM customers")->fetchColumn(), "\n";'

Upload the same file again. The count stays at 1234, because the import is an upsert.

Pitfalls

• Put a reference in the message, not the rows. The message holds a path, an offset, and a limit. Hundreds of rows in one payload make large Redis values and slow serialization.
• The file must be readable by the worker. With workers on other servers, upload to shared storage or S3 and put that key in the message.
• Chunks run in any order. A Redis set has no order, and several workers run at once. Do not rely on chunk 0 finishing before chunk 1.
• Make a chunk safe to run twice. A thrown exception retries the whole chunk. The transaction plus ON CONFLICT … DO UPDATE makes a second run harmless.
• Know when everything is done. Hermes has no batch or "all chunks finished" event. Count finished chunks yourself, for example with a Redis INCR per chunk, and act when it reaches the total.

Full source

Every file of this recipe, with its README.md, Makefile and the e2e.sh check, is in examples/csv-import on GitHub. Clone the repository, cd examples/csv-import, and run make install up, then make e2e.

FAQ

Why chunks instead of one message?

Small messages import in parallel, and a failure retries one chunk instead of the whole file.

Is a retried chunk imported twice?

No. The upsert makes a second run update the same rows.

How do I know when the import is finished?

Hermes has no batch event. Count finished chunks, for example with Redis INCR, and compare with the total.

Can workers on other servers read the file?

Only from shared storage. Upload to S3 or a shared disk and put that key in the message.

Do chunks run in order?

No. A Redis set has no order and several workers run at once.