diff --git a/phpstan-baseline.neon b/phpstan-baseline.neon
index a1ab11f..18f930b 100644
--- a/phpstan-baseline.neon
+++ b/phpstan-baseline.neon
@@ -120,11 +120,6 @@ parameters:
 			count: 1
 			path: src/Processor/EloquentProcessor.php
 
-		-
-			message: '#^Parameter \#2 \$value of method Illuminate\\Support\\Collection\<\(int\|string\),string\>\:\:put\(\) expects string, array given\.$#'
-			identifier: argument.type
-			count: 1
-			path: src/Processor/EloquentProcessor.php
 
 		-
 			message: '#^PHP files should declare strict types\.$#'
diff --git a/src/Generator/CsvDriver.php b/src/Generator/CsvDriver.php
index 4a85a4b..443cc0c 100644
--- a/src/Generator/CsvDriver.php
+++ b/src/Generator/CsvDriver.php
@@ -69,7 +69,7 @@ public function saveToStorage(
             path: $filepath,
             size: Storage::size($filepath),
             url: Storage::url($filepath),
-            count: count($processorData->getData()),
+            count: $processorData->getCount(),
             mimeType: 'text/csv',
         );
     }
@@ -101,8 +101,7 @@ public function writeCsv(ProcessorData $processorData, $stream): int
             throw new ValueError('The stream must be a resource.');
         }
 
-        $rows = $processorData->getData();
-        $columns = $this->columns($rows);
+        $columns = $processorData->getColumns();
 
         if ($columns === []) {
             return 0;
@@ -112,7 +111,7 @@ public function writeCsv(ProcessorData $processorData, $stream): int
 
         $written = 0;
 
-        foreach ($rows as $row) {
+        foreach ($processorData->rows() as $row) {
             $fields = [];
 
             foreach ($columns as $column) {
@@ -126,27 +125,6 @@ public function writeCsv(ProcessorData $processorData, $stream): int
         return $written;
     }
 
-    /**
-     * The header names every column any row has, in the order first seen, so a
-     * row whose keys differ cannot line up against the wrong header.
-     *
-     * @param array<array-key, array<array-key, mixed>> $rows
-     *
-     * @return array<int, array-key>
-     */
-    private function columns(array $rows): array
-    {
-        $columns = [];
-
-        foreach ($rows as $row) {
-            foreach (array_keys($row) as $key) {
-                $columns[$key] = true;
-            }
-        }
-
-        return array_keys($columns);
-    }
-
     /**
      * @param resource                $stream
      * @param array<array-key, mixed> $fields
diff --git a/src/Processor/EloquentProcessor.php b/src/Processor/EloquentProcessor.php
index df68a64..6c8c109 100644
--- a/src/Processor/EloquentProcessor.php
+++ b/src/Processor/EloquentProcessor.php
@@ -87,16 +87,15 @@ protected function columns(): Collection
         });
     }
 
-    protected function filterQuery(Builder $query): array
+    protected function filterQuery(Builder $query): RowStream
     {
         $columns = $this->columns();
         $allowedColumns = $columns->keys();
 
-        $items = collect();
+        $rows = new RowStream();
 
-        $query->chunk(1000, function ($data) use ($columns, $allowedColumns, $items) {
-            // Combine columns plus additional and optional fields
-            $filtered = $data->map(function ($item) use ($columns, $allowedColumns) {
+        $query->chunk(1000, function ($data) use ($columns, $allowedColumns, $rows) {
+            foreach ($data as $item) {
                 $additional = $this->additional($item);
                 $optional = $this->optional($item);
                 $additionalKeys = array_keys($additional);
@@ -106,13 +105,13 @@ protected function filterQuery(Builder $query): array
                 $attributes = $allowedColumns->isEmpty() ? [] : $this->serialiseAttributes($item);
 
                 // Merge fieldsets
-                $item = collect(array_merge($attributes, $additional));
+                $row = collect(array_merge($attributes, $additional));
 
                 // Only let through the desired columns
-                $item = $item->only($allowedColumns->concat($additionalKeys));
+                $row = $row->only($allowedColumns->concat($additionalKeys));
 
                 // Ensure some values are properly casted
-                $item = $item->map(function ($value) {
+                $row = $row->map(function ($value) {
                     if (is_bool($value)) {
                         $value = (int) $value;
                     }
@@ -120,75 +119,15 @@ protected function filterQuery(Builder $query): array
                     return strval($value);
                 });
 
-                // add optional into the item as array so we can later
-                // normalize the columns
-                $item->put('optional', $optional);
-
                 // update keys based on desired key names provided in the processor
-                $item = $item->keyBy(function ($value, $key) use ($columns) {
+                $row = $row->keyBy(function ($value, $key) use ($columns) {
                     return $columns->get($key, $key);
                 })->all();
 
-                // we return an array
-                return $item;
-            });
-
-            $items->push($filtered);
-        });
-
-        $formattedItems = $this->normalizeOptionalItems($items->flatten(1)->toArray());
-
-        return $formattedItems;
-    }
-
-    protected function getAllOptionalKeys(array $items): array
-    {
-        $optionalKeys = [];
-
-        foreach ($items as $item) {
-            $optional = $item['optional'];
-
-            if (count($optional)) {
-                foreach ($item['optional'] as $optional) {
-                    $key = key($optional);
-                    if (! in_array($key, $optionalKeys)) {
-                        $optionalKeys[] = $key;
-                    }
-                }
-            }
-        }
-
-        return $optionalKeys;
-    }
-
-    protected function normalizeOptionalItems(array $items): array
-    {
-        // get optional keys
-        $optionalKeys = $this->getAllOptionalKeys($items);
-
-        // go through each item
-        foreach ($items as $key => $item) {
-            $itemOptionals = call_user_func_array('array_merge', $items[$key]['optional']);
-
-            // for each item we wanna add each optional key
-            // and also populate value if we got one
-            foreach ($optionalKeys as $optionalArrKey => $optionalKey) {
-                $value = '';
-
-                // check if we have value for this item and populate it
-                if (isset($itemOptionals[$optionalKey])) {
-                    $value = $itemOptionals[$optionalKey];
-                }
-
-                // add the optional key and it's value
-                $items[$key][$optionalKey] = $value;
+                $rows->push($row, $optional);
             }
+        });
 
-            // unset our "temp" optional array as now we have them
-            // in the correct format
-            unset($items[$key]['optional']);
-        }
-
-        return $items;
+        return $rows;
     }
 }
diff --git a/src/Processor/ProcessorData.php b/src/Processor/ProcessorData.php
index afe66db..656694c 100644
--- a/src/Processor/ProcessorData.php
+++ b/src/Processor/ProcessorData.php
@@ -4,17 +4,55 @@
 
 class ProcessorData
 {
+    /**
+     * Rows either already built as an array, or held in a stream that can be read
+     * more than once. A one-shot iterable would be exhausted by the first reader.
+     *
+     * @param array<array-key, array<array-key, mixed>>|RowStream $data
+     */
     public function __construct(
-        private readonly array $data,
+        private readonly array|RowStream $data,
         private readonly string $type,
     ) {
     }
 
-    public function getData(): array
+    /**
+     * The rows, one at a time.
+     *
+     * @return iterable<array-key, array<array-key, mixed>>
+     */
+    public function rows(): iterable
     {
         return $this->data;
     }
 
+    public function getCount(): int
+    {
+        return count($this->data);
+    }
+
+    /**
+     * Every column any row has, in the order first seen.
+     *
+     * @return array<int, array-key>
+     */
+    public function getColumns(): array
+    {
+        if ($this->data instanceof RowStream) {
+            return $this->data->columns();
+        }
+
+        $columns = [];
+
+        foreach ($this->data as $row) {
+            foreach (array_keys($row) as $key) {
+                $columns[$key] = true;
+            }
+        }
+
+        return array_keys($columns);
+    }
+
     public function getType(): string
     {
         return $this->type;
diff --git a/src/Processor/RowStream.php b/src/Processor/RowStream.php
new file mode 100644
index 0000000..cd8263a
--- /dev/null
+++ b/src/Processor/RowStream.php
@@ -0,0 +1,189 @@
+<?php
+
+declare(strict_types=1);
+
+namespace Worksome\DataExport\Processor;
+
+use Countable;
+use Generator;
+use IteratorAggregate;
+use RuntimeException;
+use Stringable;
+
+/**
+ * The rows of an export, kept in a buffer that spills to disk rather than in memory.
+ *
+ * A row is written the moment it is built, before the full set of columns is
+ * known - an optional column only exists once some row has a value for it -
+ * so reading the rows back fills each one out to the final column list.
+ *
+ * @implements IteratorAggregate<int, array<array-key, string>>
+ */
+final class RowStream implements IteratorAggregate, Countable
+{
+    /** Rows stay in memory up to this size and go to a temp file beyond it. */
+    private const BUFFER_BYTES = 8 * 1024 * 1024;
+
+    /** @var resource */
+    private $buffer;
+
+    /** @var array<array-key, true> the fixed columns, in the order first seen */
+    private array $columns = [];
+
+    /** @var array<array-key, true> the optional columns, in the order first seen */
+    private array $optionalColumns = [];
+
+    /** How many of the fixed columns the first row had. */
+    private int $firstRowColumnCount = 0;
+
+    private int $count = 0;
+
+    public function __construct()
+    {
+        $buffer = fopen('php://temp/maxmemory:' . self::BUFFER_BYTES, 'w+b');
+
+        if ($buffer === false) {
+            throw new RuntimeException('Unable to open a buffer to hold the export rows.');
+        }
+
+        $this->buffer = $buffer;
+    }
+
+    public function __destruct()
+    {
+        if (is_resource($this->buffer)) {
+            fclose($this->buffer);
+        }
+    }
+
+    /**
+     * Values are stored as the strings the file will contain, so nothing is lost
+     * or reinterpreted on the way through the buffer.
+     *
+     * @param array<array-key, mixed>             $row      the row's fixed columns
+     * @param array<int, array<array-key, mixed>> $optional one single-entry array per optional value
+     */
+    public function push(array $row, array $optional): void
+    {
+        $fixed = [];
+
+        foreach ($row as $key => $value) {
+            $this->columns[$key] = true;
+            $fixed[$key] = $this->stringify($value);
+        }
+
+        if ($this->count === 0) {
+            $this->firstRowColumnCount = count($this->columns);
+        }
+
+        // A later optional value for the same column wins, as array_merge would have it.
+        $merged = [];
+
+        foreach ($optional as $entry) {
+            foreach ($entry as $key => $value) {
+                $this->optionalColumns[$key] = true;
+                $merged[$key] = $this->stringify($value);
+            }
+        }
+
+        // Length-framed so a value may contain anything, including newlines.
+        $payload = serialize([$fixed, $merged]);
+        $frame = strlen($payload) . "\n" . $payload;
+
+        if (fwrite($this->buffer, $frame) !== strlen($frame)) {
+            throw new RuntimeException('Unable to write an export row to the buffer.');
+        }
+
+        $this->count++;
+    }
+
+    /**
+     * The first row's columns, then the optional columns, then any column that only
+     * appears in a later row. A column keeps the first position it is given.
+     *
+     * @return array<int, array-key>
+     */
+    public function columns(): array
+    {
+        $fixed = array_keys($this->columns);
+
+        return array_values(array_unique(array_merge(
+            array_slice($fixed, 0, $this->firstRowColumnCount),
+            array_keys($this->optionalColumns),
+            array_slice($fixed, $this->firstRowColumnCount),
+        )));
+    }
+
+    public function count(): int
+    {
+        return $this->count;
+    }
+
+    /**
+     * Every row, filled out to the final column list. Can be iterated more than once.
+     *
+     * @return Generator<int, array<array-key, string>>
+     */
+    public function getIterator(): Generator
+    {
+        $columns = $this->columns();
+
+        rewind($this->buffer);
+
+        while (($header = fgets($this->buffer)) !== false) {
+            $payload = $this->read((int) $header);
+
+            /** @var array{0: array<array-key, string>, 1: array<array-key, string>} $decoded */
+            $decoded = unserialize($payload, ['allowed_classes' => false]);
+            [$fixed, $optionals] = $decoded;
+
+            $filled = [];
+
+            foreach ($columns as $column) {
+                // An optional column wins over a fixed one of the same name, and a
+                // row without a value for it gets an empty string.
+                $filled[$column] = isset($this->optionalColumns[$column])
+                    ? ($optionals[$column] ?? '')
+                    : ($fixed[$column] ?? '');
+            }
+
+            yield $filled;
+        }
+    }
+
+    private function read(int $length): string
+    {
+        $payload = '';
+
+        while (strlen($payload) < $length) {
+            $chunk = fread($this->buffer, $length - strlen($payload));
+
+            if ($chunk === false || $chunk === '') {
+                throw new RuntimeException('The export row buffer is truncated.');
+            }
+
+            $payload .= $chunk;
+        }
+
+        return $payload;
+    }
+
+    private function stringify(mixed $value): string
+    {
+        if ($value === null) {
+            return '';
+        }
+
+        if (is_string($value)) {
+            return $value;
+        }
+
+        if (is_scalar($value) || $value instanceof Stringable) {
+            return (string) $value;
+        }
+
+        throw new RuntimeException(
+            sprintf('An export value must be a string or stringable, %s given.', get_debug_type($value))
+        );
+    }
+}
diff --git a/tests/Feature/Generator/CsvDriverTest.php b/tests/Feature/Generator/CsvDriverTest.php
index 117c2ae..c5b4610 100644
--- a/tests/Feature/Generator/CsvDriverTest.php
+++ b/tests/Feature/Generator/CsvDriverTest.php
@@ -123,3 +123,18 @@
 it('returns an empty string when there is nothing to export', function () {
     expect((new CsvDriver())->exportToCsv(new ProcessorData([], 'empty')))->toBe('');
 });
+
+it('writes a row stream without loading it, with the header from its columns', function () {
+    $rows = new \Worksome\DataExport\Processor\RowStream();
+    $rows->push(['id' => '1', 'name' => 'One'], [['UK' => 'applies']]);
+    $rows->push(['id' => '2', 'name' => 'Two'], []);
+
+    $csv = (new CsvDriver())->exportToCsv(new ProcessorData($rows, 'people'));
+
+    expect($csv)->toBe("id,name,UK\r\n1,One,applies\r\n2,Two,\r\n");
+});
+
+it('rejects a stream that is not a resource, even when there is nothing to write', function () {
+    expect(fn () => (new CsvDriver())->writeCsv(new ProcessorData([], 'empty'), 'not a stream'))
+        ->toThrow(\ValueError::class);
+});
diff --git a/tests/Feature/Processor/EloquentProcessorTest.php b/tests/Feature/Processor/EloquentProcessorTest.php
index f6a7f63..9829eea 100644
--- a/tests/Feature/Processor/EloquentProcessorTest.php
+++ b/tests/Feature/Processor/EloquentProcessorTest.php
@@ -19,7 +19,7 @@
 
     $processedData = $processor->process($export);
 
-    $data = $processedData->getData();
+    $data = iterator_to_array($processedData->rows(), false);
 
     expect($data)->toBeEmpty();
 });
@@ -41,7 +41,7 @@
 
     $processedData = $processor->process($export);
 
-    $data = $processedData->getData();
+    $data = iterator_to_array($processedData->rows(), false);
 
     expect($data)->not->toBeEmpty();
     expect($data)->toHaveCount(2);
@@ -71,7 +71,7 @@
     $export = new Export();
     $processor = new FakeProcessorWithOptionalDriver();
     $processedData = $processor->process($export);
-    $data = $processedData->getData();
+    $data = iterator_to_array($processedData->rows(), false);
 
     foreach ($data as $item) {
         $this->assertArrayHasKey('Compliance UK', $item);
@@ -103,7 +103,7 @@
         $queries[] = $query->sql;
     });
 
-    $data = (new FakeProcessorWithRelationDriver())->process(new Export())->getData();
+    $data = iterator_to_array((new FakeProcessorWithRelationDriver())->process(new Export())->rows(), false);
 
     // One query for the users, one for the eagerly loaded teams. Serialising a
     // team would query its members on top of that.
@@ -123,7 +123,7 @@
         $queries[] = $query->sql;
     });
 
-    $data = (new FakeProcessorWithoutColumnsDriver())->process(new Export())->getData();
+    $data = iterator_to_array((new FakeProcessorWithoutColumnsDriver())->process(new Export())->rows(), false);
 
     // Only the chunk query over teams. Reading a team's attributes would run its
     // appended accessor, which queries the team's members.
@@ -141,7 +141,7 @@
         'email' => 'one@example.com',
     ]);
 
-    $data = (new FakeProcessorWithCustomToArrayDriver())->process(new Export())->getData();
+    $data = iterator_to_array((new FakeProcessorWithCustomToArrayDriver())->process(new Export())->rows(), false);
 
     // The model redacts the address when it serialises itself, so the export
     // must not fall back to the raw attribute.
diff --git a/tests/Feature/Processor/ProcessorDataTest.php b/tests/Feature/Processor/ProcessorDataTest.php
index 4027f88..9a89b29 100644
--- a/tests/Feature/Processor/ProcessorDataTest.php
+++ b/tests/Feature/Processor/ProcessorDataTest.php
@@ -9,9 +9,38 @@
         'foo' => 'Bar',
     ], 'randomstuff');
 
-    expect($processorData->getData())->toBe([
+    expect($processorData->rows())->toBe([
         'foo' => 'Bar',
     ]);
 
     expect($processorData->getType())->toBe('randomstuff');
 });
+
+it('streams rows and reports the columns and count for a row stream', function () {
+    $rows = new \Worksome\DataExport\Processor\RowStream();
+    $rows->push(['id' => '1'], [['UK' => 'applies']]);
+    $rows->push(['id' => '2'], []);
+
+    $processorData = new ProcessorData($rows, 'compliance');
+
+    expect($processorData->getColumns())->toBe(['id', 'UK'])
+        ->and($processorData->getCount())->toBe(2)
+        ->and(iterator_to_array($processorData->rows(), false))->toBe([
+            ['id' => '1', 'UK' => 'applies'],
+            ['id' => '2', 'UK' => ''],
+        ]);
+});
+
+it('derives the columns and count from a plain array of rows', function () {
+    $processorData = new ProcessorData([
+        ['a' => '1', 'b' => '2'],
+        ['b' => '3', 'c' => '4'],
+    ], 'shapes');
+
+    expect($processorData->getColumns())->toBe(['a', 'b', 'c'])
+        ->and($processorData->getCount())->toBe(2)
+        ->and($processorData->rows())->toBe([
+            ['a' => '1', 'b' => '2'],
+            ['b' => '3', 'c' => '4'],
+        ]);
+});
diff --git a/tests/Feature/Processor/RowStreamTest.php b/tests/Feature/Processor/RowStreamTest.php
new file mode 100644
index 0000000..a445636
--- /dev/null
+++ b/tests/Feature/Processor/RowStreamTest.php
@@ -0,0 +1,118 @@
+<?php
+
+namespace Worksome\DataExport\Tests\Feature\Processor;
+
+use Worksome\DataExport\Processor\RowStream;
+
+it('fills every row out to the final column list, fixed columns first', function () {
+    $rows = new RowStream();
+    $rows->push(['id' => '1', 'name' => 'One'], [['UK' => 'applies']]);
+    $rows->push(['id' => '2', 'name' => 'Two'], [['US' => 'none'], ['UK' => 'exempt']]);
+    $rows->push(['id' => '3', 'name' => 'Three'], []);
+
+    expect($rows->columns())->toBe(['id', 'name', 'UK', 'US'])
+        ->and(count($rows))->toBe(3)
+        ->and(iterator_to_array($rows, false))->toBe([
+            ['id' => '1', 'name' => 'One', 'UK' => 'applies', 'US' => ''],
+            ['id' => '2', 'name' => 'Two', 'UK' => 'exempt', 'US' => 'none'],
+            ['id' => '3', 'name' => 'Three', 'UK' => '', 'US' => ''],
+        ]);
+});
+
+it('places a column that only appears in a later row after the optional columns', function () {
+    $rows = new RowStream();
+    $rows->push(['id' => '1'], [['opt' => 'a']]);
+    $rows->push(['id' => '2', 'later' => 'b'], []);
+
+    // The first row's columns, then the optional ones, then the newcomer
+    expect($rows->columns())->toBe(['id', 'opt', 'later'])
+        ->and(iterator_to_array($rows, false))->toBe([
+            ['id' => '1', 'opt' => 'a', 'later' => ''],
+            ['id' => '2', 'opt' => '', 'later' => 'b'],
+        ]);
+});
+
+it('lets an optional value take precedence over a fixed column of the same name', function () {
+    $rows = new RowStream();
+    $rows->push(['id' => '1', 'status' => 'fixed'], [['status' => 'optional']]);
+    $rows->push(['id' => '2', 'status' => 'fixed'], []);
+
+    // The column keeps its fixed position, and a row without the optional value is blank
+    expect($rows->columns())->toBe(['id', 'status'])
+        ->and(iterator_to_array($rows, false))->toBe([
+            ['id' => '1', 'status' => 'optional'],
+            ['id' => '2', 'status' => ''],
+        ]);
+});
+
+it('can be read more than once', function () {
+    $rows = new RowStream();
+    $rows->push(['a' => '1'], []);
+
+    expect(iterator_to_array($rows, false))->toBe([['a' => '1']])
+        ->and(iterator_to_array($rows, false))->toBe([['a' => '1']]);
+});
+
+it('yields nothing and no columns when empty', function () {
+    $rows = new RowStream();
+
+    expect($rows->columns())->toBe([])
+        ->and(count($rows))->toBe(0)
+        ->and(iterator_to_array($rows, false))->toBe([]);
+});
+
+it('keeps values byte for byte, including newlines and bytes that are not valid UTF-8', function () {
+    $rows = new RowStream();
+    $rows->push(['note' => "line one\nline two"], [['raw' => "bad \xB1\xFF bytes"]]);
+
+    expect(iterator_to_array($rows, false))->toBe([
+        ['note' => "line one\nline two", 'raw' => "bad \xB1\xFF bytes"],
+    ]);
+});
+
+it('stores values as the strings the file will contain', function () {
+    $stringable = new class() implements \Stringable {
+        public function __toString(): string
+        {
+            return 'from __toString';
+        }
+    };
+
+    $rows = new RowStream();
+    $rows->push(
+        ['id' => 7, 'flag' => true, 'off' => false, 'none' => null],
+        [['label' => $stringable], ['rate' => 12.5]]
+    );
+
+    expect(iterator_to_array($rows, false))->toBe([
+        ['id' => '7', 'flag' => '1', 'off' => '', 'none' => '', 'label' => 'from __toString', 'rate' => '12.5'],
+    ]);
+});
+
+it('refuses a value it cannot turn into a string', function () {
+    $rows = new RowStream();
+
+    expect(fn () => $rows->push(['id' => ['not', 'a', 'string']], []))
+        ->toThrow(\RuntimeException::class);
+});
+
+it('does not hold the rows in memory', function () {
+    $rows = new RowStream();
+    $row = array_fill_keys(array_map(fn ($i) => "column_$i", range(1, 20)), str_repeat('x', 40));
+
+    memory_reset_peak_usage();
+    $before = memory_get_usage(true);
+    for ($i = 0; $i < 50000; $i++) {
+        $rows->push($row, [['extra' => 'y']]);
+    }
+    $count = 0;
+    foreach ($rows as $filled) {
+        $count++;
+    }
+    $grown = (memory_get_peak_usage(true) - $before) / 1048576;
+
+    // 50k rows of ~1 KB each would be ~50 MB if kept in memory. The buffer spills
+    // to disk past 8 MB, so growth stays a small constant however many rows there are.
+    expect($count)->toBe(50000)
+        ->and($grown)->toBeLessThan(24);
+});