Продвинутое использование

Raw API

Для сценариев, где не требуется преобразование между данными ClickHouse и собственными или сторонними типами данных и структурами, клиент ClickHouse Connect предоставляет методы для прямой работы с соединением ClickHouse.

Метод клиента `raw_query`

Метод Client.raw_query позволяет напрямую использовать HTTP-интерфейс запросов ClickHouse через клиентское соединение. Возвращаемое значение — необработанный объект bytes. Этот метод предоставляет удобную обёртку с привязкой параметров, обработкой ошибок, повторными попытками и управлением настройками через минимальный интерфейс:

Parameter	Type	Default	Description
`query`	str	Required	Любой допустимый запрос к ClickHouse.
`parameters`	dict or sequence	`None`	См. аргумент Parameters.
`settings`	dict	`None`	См. аргумент Settings.
`fmt`	str	`None`	Выходной формат ClickHouse. Если формат не указан, ClickHouse использует TSV.
`use_database`	bool	`True`	Использовать базу данных, настроенную в клиенте.
`external_data`	`ExternalData`	`None`	Внешний файл или бинарные данные. См. Внешние данные.
`transport_settings`	dict	`None`	HTTP-заголовки, добавляемые к этому запросу.

Обработка результирующего объекта bytes остаётся на стороне вызывающего кода. Обратите внимание, что Client.query_arrow — это лишь простая обёртка над этим методом, использующая выходной формат ClickHouse Arrow.

Метод `raw_stream` класса Client

Синхронный метод Client.raw_stream имеет тот же API, что и raw_query, но возвращает поток io.IOBase, состоящий из байтовых фрагментов. Закройте поток после завершения обработки. Для AsyncClient.raw_stream нужно использовать await; этот метод возвращает асинхронный StreamContext для работы с async with и async for.

Метод клиента `raw_insert`

Метод Client.raw_insert позволяет выполнять прямую вставку объектов bytes или генераторов объектов bytes через клиентское соединение. Поскольку он никак не обрабатывает полезную нагрузку вставки, он обеспечивает очень высокую производительность. Метод предоставляет параметры для указания настроек и формата вставки:

Параметр	Тип	По умолчанию	Описание
`table`	str	Обязательно	Простое имя таблицы или имя таблицы с указанием базы данных.
`column_names`	Sequence[str]	`None`	Имена столбцов для блока вставки. Обязательно, если `fmt` не включает имена.
`insert_block`	str, bytes, generator, or `BinaryIO`	Обязательно	Данные для вставки. Строки кодируются с использованием кодировки клиента.
`settings`	dict	`None`	См. аргумент Settings.
`fmt`	str	`None`	Входной формат ClickHouse для полезной нагрузки `insert_block`. Если формат не указан, используется `Native`.
`compression`	str	`None`	Сжатие, уже применённое к `insert_block`, например `"gzip"`, `"lz4"` или `"zstd"`.
`transport_settings`	dict	`None`	HTTP-заголовки, добавляемые к этому запросу.

Ответственность за то, чтобы insert_block соответствовал указанному формату и использовал указанный метод сжатия, лежит на вызывающей стороне. ClickHouse Connect использует такие необработанные вставки для загрузки файлов и таблиц PyArrow, делегируя их разбор ClickHouse server.

Сохранение результатов запроса в файлы

С помощью метода raw_stream можно напрямую в потоковом режиме записывать файлы из ClickHouse в локальную файловую систему. Например, если вы хотите сохранить результаты запроса в CSV-файл, можно использовать следующий фрагмент кода:

import clickhouse_connect

if __name__ == "__main__":
    client = clickhouse_connect.get_client()
    query = (
        "SELECT number, toString(number) AS number_as_str "
        "FROM system.numbers LIMIT 5"
    )
    stream = client.raw_stream(query=query, fmt="CSVWithNames")
    try:
        with open("output.csv", "wb") as file:
            for chunk in stream:
                file.write(chunk)
    finally:
        stream.close()
        client.close()

Приведённый выше код создаёт файл output.csv со следующим содержимым:

"number","number_as_str"
0,"0"
1,"1"
2,"2"
3,"3"
4,"4"

Аналогичным образом данные можно сохранять в формате TabSeparated и других форматах. Обзор всех доступных вариантов см. в разделе Форматы входных и выходных данных.

Сценарии использования в многопоточных, многопроцессных и асинхронных/работающих на цикле событий приложениях

ClickHouse Connect хорошо работает в многопоточных, многопроцессных и асинхронных приложениях, а также в приложениях, работающих на цикле событий. Вся обработка запросов и вставок происходит в одном потоке, поэтому операции в целом потокобезопасны. (Параллельная обработка некоторых операций на низком уровне — возможное улучшение в будущем, которое поможет избежать потерь производительности из-за использования одного потока, но даже в этом случае потокобезопасность сохранится.) Поскольку каждый выполняемый запрос или операция вставки хранит состояние в собственном объекте QueryContext или InsertContext соответственно, эти вспомогательные объекты не являются потокобезопасными и не должны совместно использоваться между несколькими потоками обработки. Дополнительные сведения об объектах контекста см. в разделах QueryContexts и InsertContexts. Кроме того, в приложении, где одновременно выполняются два или более запроса и/или вставки, нужно учитывать еще два момента. Первый — это clickHouse-«сеанс», связанный с запросом или вставкой, а второй — пул HTTP-соединений, используемый экземплярами клиента ClickHouse Connect.

AsyncClient

ClickHouse Connect предоставляет нативный клиент на базе aiohttp для приложений asyncio. Перед использованием установите дополнительную зависимость:

pip install "clickhouse-connect[async]"

Вызовите get_async_client с await, чтобы создать и инициализировать клиент. Методы ввода-вывода, такие как query, command и insert, являются корутинами:

import asyncio

import clickhouse_connect


async def main():
    async with await clickhouse_connect.get_async_client() as client:
        result = await client.query(
            "SELECT name FROM system.databases ORDER BY name LIMIT 1"
        )
        print(result.result_rows)


asyncio.run(main())

Асинхронный клиент поддерживает те же контракты query, insert, raw, Arrow и streaming, что и синхронный клиент. Для сетевого ввода-вывода используется aiohttp. Разбор Native-формата, ограниченный CPU, может выполняться в исполнителе, чтобы не блокировать цикл событий. Асинхронному клиенту принадлежит сеанс aiohttp, созданный в определённом цикле событий. Прежде чем перенести клиент в другой цикл событий, закройте его в исходном цикле, а затем, до отправки запросов, вызовите await client._initialize() в новом цикле. Если исходный цикл уже закрыт, вызовите в текущем цикле await client.close(), а затем await client._initialize(). Если очистка начинается только после закрытия исходного цикла, aiohttp всё равно может сообщить о незакрытом транспорте, поэтому по возможности закрывайте клиент до переноса. Асинхронных методов streaming нужно дождаться перед входом в возвращаемый контекст:

async with await client.query_rows_stream(
    "SELECT number FROM numbers(100000)"
) as stream:
    async for row in stream:
        process(row)

В отличие от синхронной фабрики, get_async_client по умолчанию отключает автоматическую генерацию идентификаторов сеанса, чтобы параллельно выполняющиеся корутины могли использовать один клиент совместно. Явный session_id или autogenerate_session_id=True следует передавать только в тех случаях, когда вам нужно состояние сеанса и вы можете избежать параллельных запросов в рамках этого сеанса.

Управление идентификаторами сеансов ClickHouse

Каждый запрос к ClickHouse выполняется в контексте ClickHouse “сеанса”. В настоящее время сеансы используются для двух целей:

Чтобы связывать определённые настройки ClickHouse с несколькими запросами (см. настройки пользователя). Команда ClickHouse SET используется для изменения настроек в рамках пользовательского сеанса.
Для отслеживания временных таблиц

По умолчанию синхронный Client использует сгенерированный идентификатор сеанса. Операторы SET и временные таблицы сохраняются между запросами от этого клиента, только если эти запросы попадают в один и тот же процесс сервера ClickHouse. Асинхронная фабрика по умолчанию не генерирует идентификатор сеанса. Состояние именованного сеанса и проверки на пересечение запросов в одном сеансе локальны для процесса, и клиент вызывает ProgrammingError, если обнаруживает локальное пересечение до отправки запроса. В ClickHouse Cloud и других развертываниях с балансировкой нагрузки не полагайтесь на фиксированный session_id как на распределённое состояние или распределённый мьютекс. Если пересечение запросов недопустимо, выстраивайте их последовательно ещё до отправки в ClickHouse. Используйте один из следующих подходов:

Создайте отдельный экземпляр Client для каждого thread/process/event handler, которому требуется изоляция сеанса. Это сохраняет состояние сеанса на уровне клиента (временные таблицы и значения SET).
Используйте уникальный session_id для каждого запроса через аргумент settings при вызове query, command или insert, если вам не требуется общее состояние сеанса.
Отключите сеансы для общего клиента, установив autogenerate_session_id=False перед созданием клиента (или передайте его напрямую в get_client).

import clickhouse_connect
from clickhouse_connect import common

common.set_setting("autogenerate_session_id", False)
client = clickhouse_connect.get_client(
    host="somehost.com",
    username="dbuser",
    password="password",
)

Либо передайте autogenerate_session_id=False напрямую в get_client(...). В этом случае ClickHouse Connect не отправляет session_id; сервер не считает отдельные запросы частью одного и того же сеанса. Временные таблицы и настройки уровня сеанса не будут сохраняться между запросами.

Настройка пула HTTP-соединений

ClickHouse Connect использует пулы соединений urllib3 для управления базовыми HTTP-соединениями с сервером. По умолчанию все экземпляры синхронного клиента в рамках одного процесса используют общий пул соединений, чего достаточно для большинства сценариев. Каждый воркер multiprocessing получает собственный локальный для процесса пул по умолчанию и повторно использует его для всех клиентов, созданных в этом воркере. Клиент, созданный до вызова fork, сохраняет пул родительского процесса, поэтому его не следует использовать в дочернем процессе. Пул по умолчанию поддерживает до 8 HTTP Keep-Alive-соединений с каждым сервером ClickHouse, используемым приложением. Параметры сокета по умолчанию включают TCP keepalive и TCP_NODELAY. Размерами буферов отправки и приёма сокета управляет операционная система. Для крупных многопоточных приложений может быть целесообразно использовать отдельные пулы соединений. Настроенные пулы соединений можно передать в основную функцию clickhouse_connect.get_client через именованный аргумент pool_mgr:

import clickhouse_connect
from clickhouse_connect.driver import httputil

big_pool_mgr = httputil.get_pool_manager(maxsize=16, num_pools=12)

client1 = clickhouse_connect.get_client(pool_mgr=big_pool_mgr)
client2 = clickhouse_connect.get_client(pool_mgr=big_pool_mgr)

Клиенты могут использовать общий менеджер пула, либо каждый клиент может использовать отдельный менеджер. Подробнее см. в документации urllib3 по PoolManager. Чтобы задать параметры сокета, передайте socket_options в httputil.get_pool_manager или httputil.get_pool_manager_options. Переданное значение полностью заменяет список по умолчанию, включая параметры keepalive и TCP_NODELAY. Чтобы не задавать явных параметров сокета, передайте [] или None. Асинхронный клиент использует собственный пул aiohttp вместо urllib3. Настройте его с помощью connector_limit, connector_limit_per_host и keepalive_timeout в get_async_client. Вызов await async_client.close_connections() пересоздаёт пул, не прерывая выполняющиеся запросы. Для асинхронных запросов и вставок ожидание свободного слота в пуле не ограничено тайм-аутом. Чтобы освободить занятые слоты пула, полностью считывайте или закрывайте потоковые ответы. Отсчёт connect_timeout начинается, когда слот становится доступен, и охватывает разрешение DNS-имён, установку TCP- и TLS-соединения, а также согласование с прокси. send_receive_timeout ограничивает время чтения из сокета. Чтобы ограничить время выполнения всей операции, включая ожидание пула, используйте asyncio.wait_for, например await asyncio.wait_for(client.query("SELECT 13"), timeout=30).

Главная

JavaScript

Python

Go

Java

C++

C

Сторонние клиенты

Продвинутое использование

Raw API

Метод клиента `raw_query`

Метод `raw_stream` класса Client

Метод клиента `raw_insert`

Сохранение результатов запроса в файлы

Сценарии использования в многопоточных, многопроцессных и асинхронных/работающих на цикле событий приложениях

AsyncClient

Управление идентификаторами сеансов ClickHouse

Настройка пула HTTP-соединений

​Raw API

​Метод клиента raw_query

​Метод raw_stream класса Client

​Метод клиента raw_insert

​Сохранение результатов запроса в файлы

​Сценарии использования в многопоточных, многопроцессных и асинхронных/работающих на цикле событий приложениях

​AsyncClient

​Управление идентификаторами сеансов ClickHouse

​Настройка пула HTTP-соединений

Raw API

Метод клиента `raw_query`

Метод `raw_stream` класса Client

Метод клиента `raw_insert`

Сохранение результатов запроса в файлы

Сценарии использования в многопоточных, многопроцессных и асинхронных/работающих на цикле событий приложениях

AsyncClient

Управление идентификаторами сеансов ClickHouse

Настройка пула HTTP-соединений