ArrowStream

ArrowStream — это формат Apache Arrow, работающий в «потоковом режиме». Он предназначен для обработки потоковых данных в памяти.

В примере ниже мы используем набор данных forex, доступный в Песочнице ClickHouse. К нему можно подключиться удалённо с помощью clickhouse-client, используя хост sql-clickhouse.clickhouse.com и пользователя demo (без пароля). Таблица forex находится в базе данных forex, поэтому мы выбираем её в качестве базы данных по умолчанию:

clickhouse-client --secure --host sql-clickhouse.clickhouse.com --user demo --database forex

Таблица forex хранит курсы валют. Мы можем оценить её размер и степень сжатия на диске, выполнив запрос к system.columns:

SELECT
    table,
    formatReadableSize(sum(data_compressed_bytes)) AS compressed_size,
    formatReadableSize(sum(data_uncompressed_bytes)) AS uncompressed_size,
    sum(data_compressed_bytes) / sum(data_uncompressed_bytes) AS compression_ratio
FROM system.columns
WHERE (database = 'forex') AND (table = 'forex')
GROUP BY table
ORDER BY table ASC

   ┌─table─┬─compressed_size─┬─uncompressed_size─┬───compression_ratio─┐
1. │ forex │ 63.69 GiB       │ 280.48 GiB        │ 0.22708227109363446 │
   └───────┴─────────────────┴───────────────────┴─────────────────────┘

В отличие от формата Arrow в режиме "file mode", которому для чтения нужен результат целиком, ArrowStream передаётся как последовательность батчей записей, которые потребитель может читать по мере их поступления. Благодаря этому он хорошо подходит для стриминга результата запроса напрямую в инструмент визуализации или аналитики без предварительной материализации всего набора данных.

Чтобы передавать результат потоком, отправьте запрос через HTTP-интерфейс ClickHouse с помощью запроса POST и считывайте ответ как поток Arrow. Мы отключаем сжатие вывода Arrow с помощью настройки output_format_arrow_compression_method, чтобы потребители могли декодировать батчи напрямую по мере получения.

Вывод ArrowStream представляет собой необработанные двоичные данные, поэтому вместо вывода в терминал мы направляем его в потребитель. Поток является самоописывающимся (он содержит собственную схему), поэтому здесь мы направляем его напрямую в clickhouse-local, который читает входящие батчи с помощью --input-format ArrowStream и выполняет к ним запросы как к таблице. Таблица forex большая, поэтому мы ограничиваем удалённый запрос с помощью предиката WHERE и LIMIT, чтобы этот пример оставался компактным:

curl "https://sql-clickhouse.clickhouse.com:8443/?user=demo&database=forex" \
    --data-binary "
        SELECT
            concat(base, '.', quote) AS base_quote,
            datetime AS last_update,
            CAST(bid, 'Float32') AS bid,
            CAST(ask, 'Float32') AS ask,
            ask - bid AS spread
        FROM forex
        WHERE base = 'USD' AND quote = 'CHF'
        ORDER BY datetime ASC
        LIMIT 5
        FORMAT ArrowStream
        SETTINGS output_format_arrow_compression_method='none'" \
  | clickhouse-local --input-format ArrowStream \
      --query "SELECT * FROM table ORDER BY last_update ASC FORMAT PrettyCompact"

   ┌─base_quote─┬─────────────last_update─┬────bid─┬────ask─┬────────────────spread─┐
1. │ USD.CHF    │ 2000-05-30 17:23:44.000 │  1.688 │ 1.6885 │ 0.0005000829696655273 │
2. │ USD.CHF    │ 2000-05-30 17:23:46.000 │ 1.6885 │  1.689 │ 0.0004999637603759766 │
3. │ USD.CHF    │ 2000-05-30 17:23:48.000 │ 1.6886 │ 1.6891 │ 0.0005000829696655273 │
4. │ USD.CHF    │ 2000-05-30 17:23:49.000 │ 1.6888 │ 1.6893 │ 0.0004999637603759766 │
5. │ USD.CHF    │ 2000-05-30 17:24:45.000 │  1.689 │ 1.6895 │ 0.0004999637603759766 │
   └────────────┴─────────────────────────┴────────┴────────┴───────────────────────┘

Тот же поток может инкрементально обрабатываться любым клиентом с поддержкой Arrow, который считывает его батч за батчем, а не буферизует результат целиком. Например, при использовании библиотеки Apache Arrow для JavaScript объект RecordBatchReader возвращает каждый батч записей сразу после его получения от server:

const reader = await RecordBatchReader.from(response);
await reader.open();
for await (const recordBatch of reader) {
    const batchTable = new Table(recordBatch);
    const ipcStream = tableToIPC(batchTable, 'stream');
    const bytes = new Uint8Array(ipcStream);
    table.update(bytes);
}

Полное пошаговое руководство по потоковой передаче данных ArrowStream из ClickHouse в визуализацию в реальном времени с Perspective см. в записи блога Потоковая визуализация в реальном времени с ClickHouse, Apache Arrow и Perspective.

ArrowStream использует те же настройки формата, что и Arrow.