PHP generatori: obrađujte velike skupove podataka bez trošenja cijelog RAM-a

PHP Generators: Stream Large Datasets Without Eating All Your RAM

PHP generatori: obrađujte velike skupove podataka bez trošenja cijelog RAM-a

Mihajlo
9. kolovoza 2026.
Vodiči

Generator ne čini velik skup podataka malim. On skup podataka čini inkrementalnim: jedan zapis ulazi u memoriju, jedan se zapis provjerava, a jedan izlazi. Ta razlika sprječava da izvoz od više gigabajta preraste u hitnu promjenu ograničenja memorije.

Ovaj vodič izrađuje produkcijski orijentiranu PHP 8.3 naredbu koja čita JSON razdvojen novim retcima, provjerava i normalizira svaki zapis te zapisuje NDJSON na standardni izlaz. Može prosljeđivati podatke datoteci, kompresoru ili drugom procesu uz održavanje ograničene memorije i prirodnog povratnog pritiska.

Što izrađujemo

Izvoznik prihvaća lokalnu NDJSON datoteku koja sadrži zapise o korisnicima. Svaki ulazni redak mora biti JSON objekt s poljima niza pod nazivima id, email i created_at. Dodatna polja se odbacuju, čime se sprječava da proširenje uzvodne sheme otkrije neočekivane podatke.

Put podataka je namjerno sinkron:

Generator čita jedan ograničeni redak.
PHP dekodira, provjerava i normalizira taj zapis.
Pisač u potpunosti ispisuje dobiveni JSON redak.
Tek tada generator nastavlja i čita drugi zapis.

Ako je izlaz cijev, a njezin se potrošač uspori, operativni sustav na kraju popunjava međuspremnik cijevi. Sljedeći blokirajući fwrite() čeka, što zaustavlja generator da povlači dodatni ulaz. To je siguran povratni pritisak bez redova, petlji anketiranja ili polja koje neprestano raste.

Upotreba memorije je konstantna u odnosu na broj zapisa, iako ostaje proporcionalna najvećem dopuštenom zapisu. Stoga primjenjujemo ograničenje od jednog megabajta po zapisu.

Preduvjeti i struktura projekta

Potrebni su vam PHP 8.3 sa standardnim funkcijama za JSON i filtere. Primjer postavljanja također koristi Bash, gzip, systemd i Linux host na kojem imate administratorske ovlasti.

ndjson-exporter/
├── bin/
│   └── export.php
├── deploy/
│   └── customer-export.service
└── src/
    ├── NdjsonWriter.php
    └── RecordStream.php

Nije potreban upravitelj paketa ni ovisnost treće strane. To zadržava model izvršavanja nedvosmisleno sinkronim i olakšava provjeru ponašanja povratnog pritiska.

Implementirajte ograničeni generator zapisa

Izradite src/RecordStream.php. Generator posjeduje ulazni rukovatelj i zatvara ga u bloku finally, uključujući slučajeve kada provjera ne uspije ili se potrošač zaustavi ranije.

<?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('Ulaz nije čitljiva lokalna datoteka: %s', $this->path)
            );
        }

        $handle = @fopen($this->path, 'rb');

        if ($handle === false) {
            throw new RuntimeException(
                sprintf('Nije moguće otvoriti ulaz: %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('Zapis %d premašuje ograničenje u bajtovima', $recordNumber)
                    );
                }

                if ($payload === '') {
                    throw new RuntimeException(
                        sprintf('Zapis %d je prazan', $recordNumber)
                    );
                }

                try {
                    $decoded = json_decode(
                        $payload,
                        false,
                        512,
                        JSON_THROW_ON_ERROR
                    );
                } catch (JsonException $exception) {
                    throw new RuntimeException(
                        sprintf(
                            'Zapis %d sadrži neispravan JSON: %s',
                            $recordNumber,
                            $exception->getMessage()
                        ),
                        0,
                        $exception
                    );
                }

                if (!$decoded instanceof stdClass) {
                    throw new RuntimeException(
                        sprintf('Zapis %d mora biti JSON objekt', $recordNumber)
                    );
                }

                yield $recordNumber => $this->normalize(
                    get_object_vars($decoded),
                    $recordNumber
                );
            }

            if (!feof($handle)) {
                throw new RuntimeException('Ulazni tok nije uspio prije EOF-a');
            }
        } 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(
                        'Zapis %d zahtijeva polje niza "%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('Zapis %d ima neispravan id', $recordNumber)
            );
        }

        if (filter_var($email, FILTER_VALIDATE_EMAIL) === false) {
            throw new RuntimeException(
                sprintf('Zapis %d ima neispravnu adresu e-pošte', $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(
                    'Zapis %d ima nekanonsku vrijednost created_at',
                    $recordNumber
                )
            );
        }

        return [
            'id' => $id,
            'email' => $email,
            'created_at' => $date->format(DATE_ATOM),
        ];
    }
}

Duljina proslijeđena funkciji fgets() ostavlja prostor za otkrivanje zapisa koji premašuje ograničenje. Završni redak bez znaka novog retka ostaje valjan, ali prevelik se redak odbacuje prije nego što JSON dekodiranje može povećati njegov trošak memorije.

Potpuno zapišite svaki zapis

Jedan poziv fwrite() nije zajamčeno dovoljan da potroši cijeli niz. Pisač mora napredovati kroz djelomična pisanja te i false i zapis od nula bajtova tretirati kao neuspjehe.

Izradite 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(
                        'Nije moguće kodirati zapis %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('Nije moguće isprazniti izlazni tok');
        }

        $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(
                    'Izlaz nije uspio ili je nizvodni potrošač zatvorio cijev'
                );
            }

            $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);
    }
}

Napredak ide na standardnu pogrešku, nikada na standardni izlaz, tako da telemetrija ne može oštetiti NDJSON tok. Neispravan kompresor ili zatvorena cijev uzrokuju izlaz izvoznika koji nije nula umjesto tiho skraćenog uspjeha.

Sastavite naredbu

Izradite 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("Upotreba: 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));

Naredba prihvaća samo običnu, čitljivu lokalnu datoteku. Time se isključuju PHP URL omotači i slučajna mrežna čitanja. Također čuva identifikatore kao nizove, pa vodeće nule ostaju u izvozu.

Testirajte putanje uspjeha i neuspjeha

Pokrenite ovaj integracijski test iz korijena projekta u Bashu. Koristi jedinstveni privremeni direktorij, provjerava točan izlaz, provjerava broj redaka i potvrđuje da neispravan ulaz ne uspijeva.

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' 'Očekivalo se da neispravan ulaz ne uspije' >&2
    exit 1
fi

printf 'Testovi su prošli; privremene datoteke ostaju u %s\n' "$test_dir"

Za veliku testnu datoteku pregledajte vršnu rezidentnu memoriju i propusnost bez zadržavanja izlaza:

/usr/bin/time -v \
  php -d memory_limit=64M bin/export.php \
  /srv/import/customers.ndjson > /dev/null

Broj zapisa trebao bi imati mali utjecaj na vršnu memoriju. Veličina zapisa, JSON dekodiranje, režijski troškovi PHP izvođenja i konfigurirana najveća veličina retka i dalje su važni.

Postavite kao učvršćenu paketnu uslugu

Donja usluga struji u gzip. Budući da se Bash pokreće s opcijom pipefail, neuspjeh PHP-a ili gzip-a sprječava završno premještanje. Privremene i završne datoteke nalaze se na istom datotečnom sustavu, što uspješno premještanje čini atomskim za čitatelje.

Izradite deploy/customer-export.service:

[Unit]
Description=Strujanje i komprimiranje normaliziranog izvoza korisnika
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.target

Instalirajte je s izričitim vlasništvom i dozvolama. Pokrenite useradd samo kada prvi put izrađujete račun usluge.

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-pager

Usluga ne otvara mrežni slušatelj, stoga joj ne treba pravilo vatrozida. Njezino ograničenje adresnih obitelji također sprječava uobičajene internetske veze. Ulaz je samo za čitanje, izlaz je izoliran, a restriktivni umask štiti izvezene podatke. Ako se kasnije doda nizvodna isporuka, ponovno razmotrite i mrežni sandbox i postupanje s vjerodajnicama umjesto da ih olako oslabite.

Performanse, vidljivost i kompromisi

Izvoznik bilježi broj zapisa, proteklo vrijeme, propusnost i vršnu dodijeljenu memoriju. Njegov izlazni kod razlikuje pogreške upotrebe od pogrešaka obrade, dok systemd i ljuskasta cjevovodna naredba čuvaju taj status.

Kompresija može postati usko grlo. To je prihvatljivo: povratni pritisak tjera PHP da čeka umjesto da nakuplja zapise. Ako je vrijeme CPU-a važnije od veličine izlaza, odaberite odgovarajuću razinu gzip kompresije nakon mjerenja reprezentativnih podataka.

Ovaj dizajn daje prednost ispravnosti i ograničenoj memoriji u odnosu na paralelnu propusnost. Više radnika zahtijevalo bi particioniran ulaz i determinističko sastavljanje izlaza. Dodavanje asinkronog reda bez strogog kapaciteta uništilo bi središnje jamstvo o memoriji.

Ponovni pokušaji ponovno pokreću izvoz od početka. Ne dodaju sadržaj objavljenoj datoteci, a čitatelji vide ili prethodni dovršeni izvoz ili novi dovršeni izvoz. Neuspješno pokretanje može ostaviti datoteku .part, koju sljedeće pokretanje sigurno skraćuje prije pisanja.

Uobičajeni produkcijski neuspjesi

Memorija i dalje raste: potražite iterator_to_array(), zadržane zapise, neograničene skupove ili zapisivanje u dnevnik koje međuspremnički sprema izlaz.
Naredba djeluje zaglavljeno: pregledajte nizvodni proces i datotečni sustav. Blokiranje na punoj cijevi očekivani je povratni pritisak, a ne nužno zastoj.
Izlaz je skraćen: zahtijevajte pipefail i objavite sadržaj tek nakon što cijeli cjevovod uspije.
Jedan zapis iscrpljuje memoriju: zadržite ograničenje u bajtovima i ograničenje PHP memorije. Konstantna memorija ne znači da je neograničena pojedinačna vrijednost bezopasna.
Datumi se neočekivano mijenjaju: prihvatite jedan kanonski prikaz vremenske oznake i odbacite normalizirane, ali neispravne kalendarske vrijednosti.
Gašenje ostavlja djelomične podatke: prekinite cijelu procesnu grupu i nikada ne preimenujte djelomični artefakt pri neuspjehu.

Završni kontrolni popis za provjeru

Ulaz je čitljiva lokalna NDJSON datoteka s jednim objektom po retku.
Svaki je zapis ograničen prije JSON dekodiranja.
Generator istodobno daje jedan normalizirani zapis.
Pisač obrađuje djelomična pisanja i zatvaranje nizvodnog potrošača.
Dijagnostika koristi standardnu pogrešku, a podaci standardni izlaz.
Kompresijski cjevovod radi s opcijom pipefail.
Atomski se objavljuje samo dovršeni artefakt.
Dozvole usluge štite i izvorne i izvezene podatke.
Vršna memorija ostaje stabilna kako se broj zapisa povećava.
Pokretanja s neispravnim, prevelikim, prekinutim ulazom i punim diskom vraćaju neuspjeh.

Generatori su najvrjedniji kada cijeli cjevovod poštuje njihovu lijenost. Kada svaka faza obrađuje jednu ograničenu stavku i čeka da sljedeća faza završi, velike datoteke prestaju biti problem upravljanja memorijom. Rezultat nije samo domišljata iteracija; to je predvidljiv produkcijski put podataka koji se sigurno usporava, vidljivo ne uspijeva i objavljuje samo dovršen rad.