Import CSV po častiach
Cieľ: prijať nahrané CSV akejkoľvek veľkosti, hneď odpovedať a importovať ho po častiach s 500 riadkami. Viac workerov môže importovať časti paralelne a neúspešná časť sa zopakuje samostatne.
Súbory
Redis a src/Hermes.php ako v návode na uvítací e-mail. Vytvorte priečinok storage/.
json
{
"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
<?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
<?php
declare(strict_types=1);
namespace App;
use LimitIterator;
use PDO;
use SplFileObject;
use Tomaj\Hermes\Handler\RetryableHandlerInterface;
use Tomaj\Hermes\Handler\RetryTrait;
use Tomaj\Hermes\MessageInterface;
final class ImportChunkHandler implements RetryableHandlerInterface
{
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
<?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();Spustenie
bash
composer install
mkdir -p storage
docker compose up -d redis
php bin/worker.php # terminal 1
php -S localhost:8002 -t public # terminal 2Vytvorte testovací súbor s 1234 riadkami a nahrajte ho:
bash
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/text
1234 rows queued in 3 chunks.Skontrolujte výsledok:
bash
php -r '$db = new PDO("sqlite:storage/app.sqlite"); echo $db->query("SELECT COUNT(*) FROM customers")->fetchColumn(), "\n";'Nahrajte ten istý súbor znova. Počet zostane 1234, pretože import je upsert.
Úskalia
- Do správy dajte odkaz, nie riadky. Správa obsahuje cestu, offset a limit. Stovky riadkov v jednom payloade znamenajú veľké hodnoty v Redise a pomalú serializáciu.
- Worker musí vedieť súbor prečítať. Ak sú workery na iných serveroch, nahrajte súbor na zdieľané úložisko alebo do S3 a do správy dajte jeho kľúč.
- Časti bežia v ľubovoľnom poradí. Množina v Redise nemá poradie a beží viac workerov naraz. Nespoliehajte sa, že časť 0 skončí pred časťou 1.
- Časť musí byť bezpečné spustiť dvakrát. Vyhodená výnimka zopakuje celú časť. Transakcia spolu s
ON CONFLICT … DO UPDATEurobí druhý beh neškodným. - Zistite, kedy je všetko hotové. Hermes nemá dávku ani udalosť „všetky časti skončili“. Hotové časti počítajte sami, napríklad cez Redis
INCRpre každú časť, a zareagujte, keď počet dosiahne celkový súčet.
Celý zdrojový kód
Všetky súbory tohto receptu, s README.md, Makefile a kontrolou e2e.sh, sú v examples/csv-import na GitHube. Naklonujte repozitár, spustite cd examples/csv-import a make install up, potom make e2e.
Časté otázky
Prečo časti a nie jedna správa?
Malé správy sa importujú paralelne a zlyhanie zopakuje jednu časť namiesto celého súboru.
Importuje sa zopakovaná časť dvakrát?
Nie. Vďaka upsertu druhý beh aktualizuje tie isté riadky.
Ako zistím, že import skončil?
Hermes nemá udalosť dávky. Počítajte hotové časti, napríklad cez Redis INCR, a porovnajte s celkovým počtom.
Môžu súbor čítať workery na iných serveroch?
Len zo zdieľaného úložiska. Nahrajte ho do S3 alebo na zdieľaný disk a do správy dajte jeho kľúč.
Bežia časti v poradí?
Nie. Množina v Redise nemá poradie a beží viac workerov naraz.