Unified data processing for PHP.

<?php

// ⚠ not memory-safe: $rows accumulates every record
$rows = [];

$file = fopen(__DIR__ . '/orders.csv', 'r');
if ($file === false) {
    throw new \RuntimeException('Cannot open orders.csv');
}

$headers = fgetcsv($file);
if ($headers === false) {
    fclose($file);
    throw new \RuntimeException('Empty CSV');
}

while (($line = fgetcsv($file)) !== false) {
    if (count($line) !== count($headers)) {
        continue; // malformed row, skip
    }

    $row = array_combine($headers, $line);

    // drop rows missing any required field
    if ($row['id'] === '' || $row['amount'] === '' || $row['email'] === '') {
        continue;
    }

    // CSV is all strings — cast every column by hand
    $row['id'] = (int) $row['id'];
    $row['amount'] = (float) $row['amount'];
    $row['created_at'] = (new \DateTimeImmutable($row['created_at']))->format(DATE_ATOM);

    $rows[] = $row;
}

fclose($file);

// ⚠ not memory-safe: serialises the entire dataset at once
$json = json_encode($rows, JSON_THROW_ON_ERROR | JSON_PRETTY_PRINT);

// ⚠ not memory-safe: writes the whole payload in one call
file_put_contents(__DIR__ . '/orders.json', $json);

<?php

use function Flow\ETL\DSL\{data_frame, ref};
use function Flow\ETL\Adapter\CSV\from_csv;
use function Flow\ETL\Adapter\JSON\to_json;

data_frame()
    ->read(from_csv(__DIR__ . '/orders.csv'))
    ->autoCast()
    ->filter(ref('id')->isNotNull())
    ->filter(ref('amount')->isNotNull())
    ->filter(ref('email')->isNotNull())
    ->write(to_json(__DIR__ . '/orders.json'))
    ->run();