Потоковая выгрузка из базы данных в CSV

Экспорт таблицы в файл - очень типовая задача. Пока записей немного, экспорт CSV обычно пишут в несколько строк: выполнить запрос, получить массив, превратить его в текст и отправить клиенту. На сотнях тысяч строк этот подход начинает расходовать слишком много памяти. Результат запроса целиком оказывается в Node.js, а при сборке CSV появляются дополнительные строки и буферы. Несколько одновременных выгрузок могут довести процесс до OOM.

Разберём реализацию на Node.js, Fastify, Knex и PostgreSQL. CSV будем генерировать порциями, сохраняя сортировку списка (createdAt DESC). Если в вашем кейсе другая сортировка - достаточно поменять на нужную. Отдельно придётся решить вопрос с авторизацией ссылки на скачивание.

Поток вместо буфера

Файл не будем собирать целиком. В целом, сам файл получится только на компьютере пользователя после скачивания. Fastify умеет отправлять Readable прямо в HTTP-ответ, а механизм backpressure в стримах ограничивает скорость чтения источника, если получатель не успевает принимать данные. Память все равно расходуется, но лишь на то что сейчас лежит в буфере.

В качестве источника возьмем для начала такой простой генератор. Он читает данные из базы и передает их дальше в стрим:

import { Readable } from 'node:stream';

async function* generate() {
    yield 'первая строка\r\n';
    const rows = await db('records').limit(1000);
    for (const row of rows) {
        yield row.value + '\r\n';
    }
}

const stream = Readable.from(generate());

Readable.from() запрашивает следующие значения по мере чтения. При этом уже полученный из БД массив остаётся в памяти до завершения обработки порции. Перепишем теперь этот кусок, чтобы читать данные пакетно.

Читаем из базы порциями, сохраняя порядок по дате

Для длинного экспорта LIMIT ... OFFSET ... неудобен. Запрос с OFFSET 500000 должен пройти мимо большого числа записей, прежде чем вернуть очередную тысячу. Сложность при использовании подходящего индекса примерно O(OFFSET+LIMIT). А если в начале списка появляются новые строки, границы страниц смещаются: часть записей можно получить повторно или пропустить.

Для последовательного прохода удобнее keyset-пагинация, стоимость такого поиска гораздо ниже. Следующая порция начинается после последней прочитанной записи. Но сортировки только по createdAt DESC недостаточно. У нескольких строк может совпадать время создания. Курсор из одного createdAt не задаёт однозначную границу.

Добавим id вторым полем сортировки и будем хранить в курсоре пару (createdAt, id):

async function* generate(filters) {
    let cursor = null; // { createdAt, id } последней отданной строки

    for (;;) {
        const query = db('records')
            .where(filters)
            .orderBy('createdAt', 'desc')
            .orderBy('id', 'desc')
            .limit(CHUNK_SIZE);
        
        if (cursor) {
            query.where((b) =>
                b
                    .where('createdAt', '<', cursor.createdAt)
                    .orWhere((b2) =>
                        b2
                            .where('createdAt', cursor.createdAt)
                            .andWhere('id', '<', cursor.id),
                    ),
            );
        }

        const chunk = await query;
        if (!chunk.length) break;

        for (const row of chunk) {
            yield formatRow(row);
        }

        const last = chunk[chunk.length - 1];
        cursor = { createdAt: last.createdAt, id: last.id };
    }
}

Условие для следующей порции:

createdAt < :createdAt
   OR (createdAt = :createdAt AND id < :id)

При неизменных значениях ключей такая граница не теряет записи с одинаковым createdAt. Однако это ещё не гарантия консистентного снимка. Новые записи, удаления и обновления во время экспорта могут изменить набор данных. В идеале это разрешать логически, чтобы данные, входящий в экспорт не менялись. Если, например, это временные логи - то никаких проблем вообще нет.

В PostgreSQL можно записать условие короче, сравнением кортежей:

query.whereRaw('(records."createdAt", records.id) < (?, ?)', [
    cursor.createdAt,
    cursor.id,
]);

Сравнение кортежей также предполагает, что поля курсора не содержат NULL. Также, для этого запроса полезен индекс по (createdAt DESC, id DESC). Начать можно с блока в 1000 строк. Оптимальный размер зависит от ширины записей, задержки БД и количества одновременных выгрузок, примерно 1к-10к оптимальнее всего.

Наивный курсор теряет строки

Но у этого кода есть неочевидная проблема, которая вылезла на реальных данных. Курсор { createdAt: last.createdAt, id } хранит время как JS Date, а Date держит только миллисекунды. Колонка времени в PostgreSQL — микросекунды. Драйвер читает значение в Date и обрезает его. Микросекунды теряются ещё до того, как курсор уйдёт в следующий запрос.

Строка, попавшая в ту же миллисекунду, что и курсор, но с меньшими микросекундами, по-настоящему идёт после курсора. Но округлённое время курсора делает сравнение (createdAt, id) < (cursor…) неверным. Пффф. Реальное createdAt такой строки больше округлённого курсора, поэтому она не проходит ни туда и ни сюда.

Лечится тем, что время для курсора нельзя прогонять через Date. Берём его из базы текстом, в полной точности, и так же текстом возвращаем в seek, приводя обратно к типу колонки:

const query = db('records')
    .select('records.*', db.raw('records."createdAt"::text as "cursorTs"'))
    .where(filters)
    .orderBy('createdAt', 'desc')
    .orderBy('id', 'desc')
    .limit(CHUNK_SIZE);

if (cursor) {
    query.whereRaw('(records."createdAt", records.id) < (?::timestamptz, ?)', [
        cursor.createdAt,
        cursor.id,
    ]);
}

const chunk = await query;
const last = chunk[chunk.length - 1];
cursor = { createdAt: last.cursorTs, id: last.id };

Тепер строка с микросекундами (…094321) уходит в следующий запрос как есть, PostgreSQL кастует её обратно в timestamptz без потерь, и граница снова совпадает с сортировкой.

Формируем строку CSV

В CSV нельзя просто соединить значения разделителем. В самих данных встречаются кавычки, ; и переводы строк и все это важно экранировать. Такие поля заключаем в двойные кавычки, а внутренние " удваиваем.

const DELIMITER = ';';

const csvField = (value) => {
    if (value == null) return '';
    const str = value instanceof Date ? value.toISOString() : String(value);
    return /[";\r\n]/.test(str) ? `"${str.replace(/"/g, '""')}"` : str;
};

Для этого экспорта приняты следующие соглашения:

null и undefined → пустое поле. Иначе в файл попадёт строка "null", и столбец станет бесполезным для последующей обработки.
Даты — в ISO 8601. toISOString() даёт однозначный вид в UTC. Локализованное форматирование лучше оставить тому, кто откроет файл.
Разделитель — точка с запятой. Excel в локалях, где десятичный разделитель — запятая (включая русскую), по умолчанию ждёт в CSV именно ;. С запятой файл откроется одной колонкой. Если потребитель — не Excel, а скрипт, выбор свободнее, но тогда стоит зафиксировать его явно.

Строку собираем из полей в фиксированном порядке колонок:

const COLUMNS = ['email', 'status', 'createdAt'];

const formatRow = (row) =>
    COLUMNS.map((col) => csvField(row[col])).join(DELIMITER) + '\r\n';

Для заголовка и строк используем один и тот же COLUMNS. Для совместимости с Excel используем \r\n и UTF-8 BOM. BOM помогает некоторым версиям Excel правильно распознать кириллицу. Без него поведение зависит от версии и способа открытия файла:

async function* generate(filters) {
    yield '' + COLUMNS.join(DELIMITER) + '\r\n';
    // ... блоки строк
}

Отдаём поток в HTTP-ответ

Fastify принимает поток в reply.send(). Для скачивания выставим тип содержимого и Content-Disposition:

app.get('/export/download', async (req, reply) => {
    const { filename, stream } = await buildExport(req.query);

    const safeName = filename.replace(/["\r\n]/g, '');
    reply.header('Content-Disposition', `attachment; filename="${safeName}"`);
    reply.type('text/csv; charset=utf-8');

    return reply.send(stream);
});

Имя файла нельзя без проверки вставлять в HTTP-заголовок. Здесь удалены кавычки и переводы строки. В рабочем коде лучше формировать безопасное ASCII-имя либо корректный filename* для Unicode, а также ограничивать допустимые символы. Если клиент оборвал соединение, стоит отдельно проверить, прекращает ли приложение чтение из базы. Иначе генератор может продолжить выполнять запросы после отмены скачивания.

Скачивание как ссылка: проблема авторизации

Обычная ссылка на скачивание не позволяет установить произвольный заголовок Authorization. Если приложение использует Bearer-токен, а не cookie-сессию, такой запрос нужно авторизовать иначе. Альтернатива — скачать файл через fetch и Blob, но тогда браузерная сторона может начать буферизовать большой файл. Один из вариантов — отдельный короткоживущий подписанный JWT для скачивания. Сначала клиент получает его через обычный авторизованный API-запрос, затем открывает ссылку:

const token = jwt.sign(
    { type: 'export-download', userId, filters },
    secret,
    { expiresIn: '5m' },
);

Клиент инициирует нативное скачивание и наш короткий токен лежит в GET параметрах. То что он попадет куда-то в логи - не страшно, ведь он все равно живет очень мало.

const url = `${apiUrl}/export/download?token=${token}`;
const a = document.createElement('a');
a.href = url;
document.body.append(a);
a.click();
a.remove();

Пять минут ограничивают время использования ссылки, но не делают токен одноразовым. До истечения срока тот же URL можно открыть повторно. Для настоящего одноразового доступа нужен jti и серверное хранилище использованных идентификаторов (например, Redis) с атомарным погашением. Токен в URL также может попасть в логи и историю, поэтому следует избегать его записи в access-логи и настроить подходящий Referrer-Policy. Будем считать что для текущего примера все это некритично.

В JWT достаточно поместить идентификатор пользователя и параметры фильтрации. Фильтры нужно ограничить по размеру: JWT передаётся в URL, а длина URL не бесконечна.

Проверка токена и область видимости

Маршрут скачивания проверяет подпись, срок действия и назначение JWT самостоятельно:

async function buildExport(token) {
    const payload = jwt.verify(token, secret);
    if (payload.type !== 'export-download') {
        throw new Unauthorized('Invalid download token');
    }
    
    const user = await getUser(payload.userId);
    if (!user) throw new Unauthorized('Invalid download token');
    const filters = pickKnownFilters(payload.filters);
    return { filename: makeName(), stream: streamFor(filters, user) };
}

Здесь важно не перепутать проверку подписи с проверкой прав доступа. Подпись защищает содержимое токена от подмены, но не заменяет проверку прав. Пользователя загружаем по userId, а ограничения по организации и роли добавляем на сервере. Фильтры из токена не должны расширять эту область видимости. Токен скачивания нельзя принимать как сессионный. У него собственный type, и в хуке обычной авторизации токены этого типа отклоняются:

const DOWNLOAD_TYPES = ['export-download'];

// В глобальном хуке авторизации
if (DOWNLOAD_TYPES.includes(decoded.type)) {
    throw new Error('Invalid token type');
}

// И путь скачивания исключается из Bearer-проверки
if (request.url.startsWith('/export/download')) {
    return;
}

Проверку типа лучше делать централизованно для всех маршрутов, принимающих JWT. При jwt.verify() также следует явно ограничить допустимые алгоритмы и проверить нужные issuer/audience, если они используются в приложении.

Один фильтр на список и на экспорт

Если список и экспорт собирают SQL независимо, со временем в них появляются разные фильтры или проверки доступа. Удобнее иметь общий построитель запроса без пагинации, чтобы убрать лишнее дублирование:

const buildQuery = (filters, user) => {
    const q = db('records');

    if (filters.status) q.where('status', filters.status);
    if (filters.email) q.where('email', 'ilike', `%${filters.email}%`);
    
    if (!user.isAdmin) q.where('ownerId', user.orgId);

    return q;
};

Далее список добавляет свою пагинацию, а экспорт — seek-условие. Сортировку оставляем одинаковой:

// Список: обычная постраничная выборка
const page = await buildQuery(filters, user)
    .orderBy('createdAt', 'desc')
    .orderBy('id', 'desc')
    .page(current, pageSize);

// Блок экспорта: тот же порядок, но seek по курсору вместо OFFSET
const chunk = await buildQuery(filters, user)
    // курсор в полной точности — см. «Наивный курсор теряет строки»
    .select('records.*', db.raw('records."createdAt"::text as "cursorTs"'))
    .orderBy('createdAt', 'desc')
    .orderBy('id', 'desc')
    .whereRaw('(records."createdAt", records.id) < (?::timestamptz, ?)', [
        cursor.createdAt, // строка из cursorTs предыдущего блока
        cursor.id,
    ])
    .limit(CHUNK_SIZE);

В примере экспорта условие whereRaw добавляется только после получения первой порции: при cursor === null оно не нужно. Общий запрос обеспечивает одинаковые правила фильтрации, но не фиксирует состояние БД на момент открытия списка. Если нужен именно снимок данных на определённый момент, потребуется отдельная стратегия консистентности.

Что проверить перед продакшеном

Сам код генератора небольшой. Больше вопросов возникает вокруг него: что происходит при отмене скачивания, сколько памяти занимают десять одновременных экспортов, какие планы строит PostgreSQL для реальных фильтров и что делать, если строки меняются прямо во время выгрузки.

Отдельно стоит протестировать CSV на кавычках, переводах строк, Unicode и значениях, которые Excel может интерпретировать как формулы (=, +, -, @). Если файл открывают пользователи в Excel, это уже вопрос не только форматирования, но и защиты от CSV formula injection. Для такого сценария нужна отдельная политика обработки потенциально опасных значений; универсальное экранирование кавычками проблему не решает.

Потоковая передача снимает основную проблему с хранением готового файла в памяти. Собственно достигается то, что на сервере никакого файл уже нет. Файл появляется только на машине клиента при скачивании. И серверу уже не критичено весит он 1 или 100Мб.

Сколько это стоит на практике

Все это было бы совсем неинтересно без замеров. Поэтому сравним все три варианта, что были в статье: потом к keyset, поток с offset и наивный вариант. Для теста возьмем локальную базу, где уже накопилось 347k строк. Готовый CSV весит около 37мб.

Для замеров памяти каждые 25мс снимаем через process.memoryUsage() heapUsed и rss, а перед каждым прогоном дергаем global.gc() (Node запущен с --expose-gc).

Вариант	Время	peak heapUsed	peak RSS
keyset, поток	2.8 с	66 МБ	173 МБ
OFFSET, поток	14.5 с	86 МБ	222 МБ
всё в памяти	2.0 с	576 МБ	833 МБ

Замеры подтверидили все в полном обьеме:

Наивное решение ест очень много памяти
OFFSET память не ест, но медленнее по времени
keyset оправдал ожидание, он примерно как наивный метод по скорости, но на порядок меньше ест памяти

Дальше в следующей статье, интересно будет разобрать и сравнить еще два варианта: pg-copy-streams и pg-query-stream)

Потоковая выгрузка из Postgres в CSV на Node.js и Fastify