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

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

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

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

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

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

Ответственность за то, чтобы 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.

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, может выполняться в исполнителе, чтобы не блокировать цикл событий.

Асинхронных методов 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 SET используется для изменения настроек в рамках пользовательского сеанса.
• Для отслеживания временных таблиц

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

1. Создайте отдельный экземпляр Client для каждого thread/process/event handler, которому требуется изоляция сеанса. Это сохраняет состояние сеанса на уровне клиента (временные таблицы и значения SET).
2. Используйте уникальный session_id для каждого запроса через аргумент settings при вызове query, command или insert, если вам не требуется общее состояние сеанса.
3. Отключите сеансы для общего клиента, установив 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; сервер не считает отдельные запросы частью одного и того же сеанса. Временные таблицы и настройки уровня сеанса не будут сохраняться между запросами.

ClickHouse Connect использует пулы соединений urllib3 для управления базовыми HTTP-соединениями с сервером. По умолчанию все экземпляры клиента используют общий пул соединений, чего достаточно для большинства сценариев. Этот пул по умолчанию поддерживает до 8 HTTP Keep-Alive-соединений с каждым сервером ClickHouse, используемым приложением.

Для крупных многопоточных приложений может быть целесообразно использовать отдельные пулы соединений. Настроенные пулы соединений можно передать в основную функцию 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.

Асинхронный клиент использует собственный пул aiohttp вместо urllib3. Настройте его с помощью connector_limit, connector_limit_per_host и keepalive_timeout в get_async_client. Вызов await async_client.close_connections() пересоздаёт пул, не прерывая выполняющиеся запросы.