PHP Generators: Stream Large Datasets Without Eating All Your RAM
A generator does not make a large dataset small. It makes the dataset incremental: one record enters memory, one record is validated, and one record leaves. That distinction is what keeps a multi-gigabyte export from turning into an emergency memory-limit change.
This tutorial builds a production-oriented PHP 8.3 command that reads newline-delimited JSON, validates and normalizes each record, and writes NDJSON to standard output. It can feed a file, compressor, or another process while maintaining bounded memory and natural backpressure.
What we are building
The exporter accepts a local NDJSON file containing customer records. Every input line must be a JSON object with string fields named id, email, and created_at. Extra fields are discarded, preventing an upstream schema expansion from leaking unexpected data.
The data path is deliberately synchronous:
- The generator reads one bounded line.
- PHP decodes, validates, and normalizes that record.
- The writer completely emits the resulting JSON line.
- Only then does the generator resume and read another record.
If the output is a pipe and its consumer slows down, the operating system eventually fills the pipe buffer. The next blocking fwrite() waits, which stops the generator from pulling more input. That is safe backpressure without queues, polling loops, or an ever-growing array.
Memory use is constant with respect to the number of records, although it remains proportional to the largest permitted record. We therefore enforce a one-megabyte per-record limit.
Prerequisites and project structure
You need PHP 8.3 with the standard JSON and filter functionality enabled. The deployment example also uses Bash, gzip, systemd, and a Linux host on which you have administrative privileges.
ndjson-exporter/
├── bin/
│ └── export.php
├── deploy/
│ └── customer-export.service
└── src/
├── NdjsonWriter.php
└── RecordStream.phpNo package manager or third-party dependency is required. This keeps the execution model unambiguously synchronous and makes the backpressure behavior easy to inspect.
Implement the bounded record generator
Create src/RecordStream.php. The generator owns the input handle and closes it in a finally block, including when validation fails or the consumer stops early.
<?php
declare(strict_types=1);
final class RecordStream
{
private const MAX_RECORD_BYTES = 1_048_576;
public function __construct(
private readonly string $path,
) {
}
/**
* @return Generator<int, array{id: string, email: string, created_at: string}>
*/
public function records(): Generator
{
if (!is_file($this->path) || !is_readable($this->path)) {
throw new RuntimeException(
sprintf('Input is not a readable local file: %s', $this->path)
);
}
$handle = @fopen($this->path, 'rb');
if ($handle === false) {
throw new RuntimeException(
sprintf('Could not open input: %s', $this->path)
);
}
try {
$recordNumber = 0;
while (
($line = fgets($handle, self::MAX_RECORD_BYTES + 2)) !== false
) {
$recordNumber++;
$hasNewline = str_ends_with($line, "\n");
$payload = rtrim($line, "\r\n");
if (
(!$hasNewline && !feof($handle))
|| strlen($payload) > self::MAX_RECORD_BYTES
) {
throw new RuntimeException(
sprintf('Record %d exceeds the byte limit', $recordNumber)
);
}
if ($payload === '') {
throw new RuntimeException(
sprintf('Record %d is empty', $recordNumber)
);
}
try {
$decoded = json_decode(
$payload,
false,
512,
JSON_THROW_ON_ERROR
);
} catch (JsonException $exception) {
throw new RuntimeException(
sprintf(
'Record %d contains invalid JSON: %s',
$recordNumber,
$exception->getMessage()
),
0,
$exception
);
}
if (!$decoded instanceof stdClass) {
throw new RuntimeException(
sprintf('Record %d must be a JSON object', $recordNumber)
);
}
yield $recordNumber => $this->normalize(
get_object_vars($decoded),
$recordNumber
);
}
if (!feof($handle)) {
throw new RuntimeException('The input stream failed before EOF');
}
} finally {
fclose($handle);
}
}
/**
* @param array<string, mixed> $record
* @return array{id: string, email: string, created_at: string}
*/
private function normalize(array $record, int $recordNumber): array
{
foreach (['id', 'email', 'created_at'] as $field) {
if (!array_key_exists($field, $record) || !is_string($record[$field])) {
throw new RuntimeException(
sprintf(
'Record %d requires string field "%s"',
$recordNumber,
$field
)
);
}
}
$id = trim($record['id']);
$email = trim($record['email']);
$createdAt = trim($record['created_at']);
if ($id === '' || !ctype_digit($id)) {
throw new RuntimeException(
sprintf('Record %d has an invalid id', $recordNumber)
);
}
if (filter_var($email, FILTER_VALIDATE_EMAIL) === false) {
throw new RuntimeException(
sprintf('Record %d has an invalid email address', $recordNumber)
);
}
$date = DateTimeImmutable::createFromFormat(
'!Y-m-d\TH:i:sP',
$createdAt
);
if (
$date === false
|| $date->format('Y-m-d\TH:i:sP') !== $createdAt
) {
throw new RuntimeException(
sprintf(
'Record %d has a non-canonical created_at value',
$recordNumber
)
);
}
return [
'id' => $id,
'email' => $email,
'created_at' => $date->format(DATE_ATOM),
];
}
}The length passed to fgets() leaves room to observe a record that exceeds the limit. A final line without a newline remains valid, but an oversized line is rejected before JSON decoding can amplify its memory cost.
Write every record completely
A single fwrite() is not guaranteed to consume the entire string. The writer must advance through partial writes and treat both false and a zero-byte write as failures.
Create src/NdjsonWriter.php:
<?php
declare(strict_types=1);
final class NdjsonWriter
{
/**
* @param iterable<int, array<string, string>> $records
* @param resource $output
*/
public function write(
iterable $records,
$output,
int $reportEvery = 100_000
): int {
$count = 0;
$startedAt = hrtime(true);
foreach ($records as $recordNumber => $record) {
try {
$json = json_encode(
$record,
JSON_THROW_ON_ERROR | JSON_UNESCAPED_SLASHES
);
} catch (JsonException $exception) {
throw new RuntimeException(
sprintf(
'Could not encode record %d: %s',
$recordNumber,
$exception->getMessage()
),
0,
$exception
);
}
$this->writeAll($output, $json . "\n");
$count++;
if ($reportEvery > 0 && $count % $reportEvery === 0) {
$this->report($count, $startedAt);
}
}
if (!fflush($output)) {
throw new RuntimeException('Could not flush the output stream');
}
$this->report($count, $startedAt);
return $count;
}
/**
* @param resource $output
*/
private function writeAll($output, string $bytes): void
{
$offset = 0;
$length = strlen($bytes);
while ($offset < $length) {
$written = @fwrite($output, substr($bytes, $offset));
if ($written === false || $written === 0) {
throw new RuntimeException(
'Output failed or the downstream consumer closed the pipe'
);
}
$offset += $written;
}
}
private function report(int $count, int $startedAt): void
{
$seconds = max((hrtime(true) - $startedAt) / 1_000_000_000, 0.001);
$message = sprintf(
"records=%d elapsed_seconds=%.3f records_per_second=%.1f peak_bytes=%d\n",
$count,
$seconds,
$count / $seconds,
memory_get_peak_usage(true)
);
@fwrite(STDERR, $message);
}
}Progress goes to standard error, never standard output, so telemetry cannot corrupt the NDJSON stream. A broken compressor or closed pipe produces a nonzero exporter exit instead of a silently truncated success.
Assemble the command
Create bin/export.php:
<?php
declare(strict_types=1);
require dirname(__DIR__) . '/src/RecordStream.php';
require dirname(__DIR__) . '/src/NdjsonWriter.php';
function main(array $arguments): int
{
if (count($arguments) !== 2) {
@fwrite(
STDERR,
sprintf("Usage: php %s /path/to/customers.ndjson\n", $arguments[0])
);
return 64;
}
try {
$stream = new RecordStream($arguments[1]);
$writer = new NdjsonWriter();
$writer->write($stream->records(), STDOUT);
return 0;
} catch (Throwable $exception) {
@fwrite(STDERR, 'export_error=' . $exception->getMessage() . "\n");
return 1;
}
}
exit(main($argv));The command accepts only a regular, readable local file. That excludes PHP URL wrappers and accidental network reads. It also preserves identifiers as strings, so leading zeroes survive the export.
Test success and failure paths
Run this integration test from the project root under Bash. It uses a unique temporary directory, verifies exact output, checks the line count, and confirms malformed input fails.
set -euo pipefail
test_dir="$(mktemp -d)"
cat > "$test_dir/input.ndjson" <<'EOF'
{"id":"001","email":"[email protected]","created_at":"2026-01-01T00:00:00+00:00","internal_note":"discard me"}
{"id":"002","email":"[email protected]","created_at":"2026-01-02T12:30:00+01:00"}
EOF
php -d memory_limit=64M bin/export.php \
"$test_dir/input.ndjson" > "$test_dir/actual.ndjson"
diff -u \
<(printf '%s\n' \
'{"id":"001","email":"[email protected]","created_at":"2026-01-01T00:00:00+00:00"}' \
'{"id":"002","email":"[email protected]","created_at":"2026-01-02T12:30:00+01:00"}') \
"$test_dir/actual.ndjson"
test "$(wc -l < "$test_dir/actual.ndjson")" -eq 2
printf '%s\n' '{"id":"003","email":"not-an-email","created_at":"2026-01-03T00:00:00+00:00"}' \
> "$test_dir/invalid.ndjson"
if php bin/export.php "$test_dir/invalid.ndjson" \
> "$test_dir/rejected.ndjson"; then
printf '%s\n' 'Expected invalid input to fail' >&2
exit 1
fi
printf 'Tests passed; temporary files remain at %s\n' "$test_dir"For a large fixture, inspect peak resident memory and throughput without retaining output:
/usr/bin/time -v \
php -d memory_limit=64M bin/export.php \
/srv/import/customers.ndjson > /dev/nullRecord count should have little effect on peak memory. Record size, JSON decoding, PHP runtime overhead, and the configured maximum line size still matter.
Deploy as a hardened batch service
The service below streams into gzip. Because Bash runs with pipefail, a PHP or gzip failure prevents the final move. The temporary and final files reside on the same filesystem, making the successful move atomic for readers.
Create deploy/customer-export.service:
[Unit]
Description=Stream and compress the normalized customer export
After=local-fs.target
[Service]
Type=oneshot
User=customer-export
Group=customer-export
UMask=0077
ExecStart=/bin/bash -o pipefail -c '/usr/bin/php -d memory_limit=64M /opt/customer-export/bin/export.php /srv/import/customers.ndjson | /usr/bin/gzip -c > /srv/export/customers.ndjson.gz.part && /usr/bin/mv -f /srv/export/customers.ndjson.gz.part /srv/export/customers.ndjson.gz'
TimeoutStopSec=30s
KillMode=control-group
NoNewPrivileges=yes
PrivateDevices=yes
PrivateTmp=yes
ProtectHome=yes
ProtectSystem=strict
ReadOnlyPaths=/opt/customer-export /srv/import
ReadWritePaths=/srv/export
RestrictAddressFamilies=AF_UNIX
RestrictSUIDSGID=yes
LockPersonality=yes
MemoryDenyWriteExecute=yes
MemoryMax=128M
[Install]
WantedBy=multi-user.targetInstall it with explicit ownership and permissions. Run useradd only when creating the service account for the first time.
sudo useradd --system \
--home-dir /nonexistent \
--shell /usr/sbin/nologin \
customer-export
sudo install -d -o root -g root -m 0755 /opt/customer-export
sudo install -d -o root -g root -m 0755 /opt/customer-export/bin
sudo install -d -o root -g root -m 0755 /opt/customer-export/src
sudo install -d -o root -g customer-export -m 0750 /srv/import
sudo install -d -o customer-export -g customer-export -m 0750 /srv/export
sudo install -o root -g root -m 0644 \
src/RecordStream.php /opt/customer-export/src/RecordStream.php
sudo install -o root -g root -m 0644 \
src/NdjsonWriter.php /opt/customer-export/src/NdjsonWriter.php
sudo install -o root -g root -m 0644 \
bin/export.php /opt/customer-export/bin/export.php
sudo install -o root -g root -m 0644 \
deploy/customer-export.service \
/etc/systemd/system/customer-export.service
sudo install -o root -g customer-export -m 0640 \
customers.ndjson /srv/import/customers.ndjson.next
sudo mv -f \
/srv/import/customers.ndjson.next \
/srv/import/customers.ndjson
sudo systemctl daemon-reload
sudo systemctl start customer-export.service
sudo systemctl status customer-export.service
sudo journalctl -u customer-export.service --no-pagerThe service opens no network listener, so it needs no firewall rule. Its address-family restriction also prevents ordinary internet connections. Input is read-only, output is isolated, and the restrictive umask protects exported data. If downstream delivery is added later, revisit both the network sandbox and credential handling rather than weakening them casually.
Performance, observability, and trade-offs
The exporter logs record count, elapsed time, throughput, and peak allocated memory. Its exit code distinguishes usage errors from processing failures, while systemd and the shell pipeline preserve that status.
Compression may become the bottleneck. That is acceptable: backpressure makes PHP wait instead of accumulating records. If CPU time matters more than output size, select an appropriate gzip compression level after measuring representative data.
This design favors correctness and bounded memory over parallel throughput. Multiple workers would require partitioned input and deterministic output assembly. Adding an asynchronous queue without a strict capacity would destroy the central memory guarantee.
Retries restart the export from the beginning. They do not append to the published file, and readers see either the previous completed export or the new completed export. A failed run can leave a .part file, which the next run safely truncates before writing.
Common production failures
- Memory still grows: look for
iterator_to_array(), retained records, unbounded batches, or logging that buffers output. - The command appears stuck: inspect the downstream process and filesystem. Blocking on a full pipe is expected backpressure, not necessarily a deadlock.
- Output is truncated: require
pipefailand publish only after the complete pipeline succeeds. - One record exhausts memory: retain the byte limit and the PHP memory limit. Constant memory does not mean an unbounded individual value is harmless.
- Dates change unexpectedly: accept one canonical timestamp representation and reject normalized-but-invalid calendar values.
- Shutdown leaves partial data: terminate the whole process group and never rename the partial artifact on failure.
Final verification checklist
- The input is a readable local NDJSON file with one object per line.
- Every record is bounded before JSON decoding.
- The generator yields one normalized record at a time.
- The writer handles partial writes and downstream closure.
- Diagnostics use standard error and data uses standard output.
- The compression pipeline runs with
pipefail. - Only a completed artifact is atomically published.
- Service permissions protect both source and exported data.
- Peak memory remains stable as the number of records increases.
- Malformed, oversized, interrupted, and disk-full runs return failure.
Generators are most valuable when the entire pipeline respects their laziness. Once every stage processes one bounded item and waits for the next stage to finish, large files stop being a memory-management problem. The result is not merely clever iteration; it is a predictable production data path that slows down safely, fails visibly, and publishes only complete work.
