> ## Documentation Index
> Fetch the complete documentation index at: https://private-7c7dfe99-parallel-read-in-order-multi-part.mintlify.site/llms.txt
> Use this file to discover all available pages before exploring further.
> Продвинутое использование с ClickHouse Connect
# Продвинутое использование
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](/ru/integrations/language-clients/python/driver-api#parameters-argument). |
| `settings` | dict | `None` | См. [аргумент Settings](/ru/integrations/language-clients/python/driver-api#settings-argument-1). |
| `fmt` | str | `None` | Выходной формат ClickHouse. Если формат не указан, ClickHouse использует TSV. |
| `use_database` | bool | `True` | Использовать базу данных, настроенную в клиенте. |
| `external_data` | `ExternalData` | `None` | Внешний файл или бинарные данные. См. [Внешние данные](/ru/integrations/language-clients/python/advanced-querying#external-data). |
| `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](/ru/integrations/language-clients/python/driver-api#settings-argument-1). |
| `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-файл, можно использовать следующий фрагмент кода:
```python theme={null}
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` со следующим содержимым:
```csv theme={null}
"number","number_as_str"
0,"0"
1,"1"
2,"2"
3,"3"
4,"4"
```
Аналогичным образом данные можно сохранять в формате [TabSeparated](/ru/reference/formats/TabSeparated/TabSeparated) и других форматах. Обзор всех доступных вариантов см. в разделе [Форматы входных и выходных данных](/ru/reference/formats).
Сценарии использования в многопоточных, многопроцессных и асинхронных/работающих на цикле событий приложениях
ClickHouse Connect хорошо работает в многопоточных, многопроцессных и асинхронных приложениях, а также в приложениях, работающих на цикле событий. Вся обработка запросов и вставок происходит в одном потоке, поэтому операции в целом потокобезопасны. (Параллельная обработка некоторых операций на низком уровне — возможное улучшение в будущем, которое поможет избежать потерь производительности из-за использования одного потока, но даже в этом случае потокобезопасность сохранится.)
Поскольку каждый выполняемый запрос или операция вставки хранит состояние в собственном объекте `QueryContext` или `InsertContext` соответственно, эти вспомогательные объекты не являются потокобезопасными и не должны совместно использоваться между несколькими потоками обработки. Дополнительные сведения об объектах контекста см. в разделах [QueryContexts](/ru/integrations/language-clients/python/advanced-querying#querycontexts) и [InsertContexts](/ru/integrations/language-clients/python/advanced-inserting#insertcontexts).
Кроме того, в приложении, где одновременно выполняются два или более запроса и/или вставки, нужно учитывать еще два момента. Первый — это clickHouse-«сеанс», связанный с запросом или вставкой, а второй — пул HTTP-соединений, используемый экземплярами клиента ClickHouse Connect.
AsyncClient
ClickHouse Connect предоставляет нативный клиент на базе aiohttp для приложений asyncio. Перед использованием установите дополнительную зависимость:
```bash theme={null}
pip install "clickhouse-connect[async]"
```
Вызовите `get_async_client` с `await`, чтобы создать и инициализировать клиент. Методы ввода-вывода, такие как `query`, `command` и `insert`, являются корутинами:
```python theme={null}
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 нужно дождаться перед входом в возвращаемый контекст:
```python theme={null}
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 с несколькими запросами (см. [настройки пользователя](/ru/reference/settings/session-settings)). Команда ClickHouse `SET` используется для изменения настроек в рамках пользовательского сеанса.
* Для отслеживания [временных таблиц](/ru/reference/statements/create/table#temporary-tables)
По умолчанию синхронный `Client` использует сгенерированный идентификатор сеанса. Операторы `SET` и временные таблицы сохраняются между запросами от этого клиента, только если эти запросы попадают в один и тот же процесс сервера ClickHouse. Асинхронная фабрика по умолчанию не генерирует идентификатор сеанса. Состояние именованного сеанса и проверки на пересечение запросов в одном сеансе локальны для процесса, и клиент вызывает `ProgrammingError`, если обнаруживает локальное пересечение до отправки запроса. В ClickHouse Cloud и других развертываниях с балансировкой нагрузки не полагайтесь на фиксированный `session_id` как на распределённое состояние или распределённый мьютекс. Если пересечение запросов недопустимо, выстраивайте их последовательно ещё до отправки в ClickHouse. Используйте один из следующих подходов:
1. Создайте отдельный экземпляр `Client` для каждого thread/process/event handler, которому требуется изоляция сеанса. Это сохраняет состояние сеанса на уровне клиента (временные таблицы и значения `SET`).
2. Используйте уникальный `session_id` для каждого запроса через аргумент `settings` при вызове `query`, `command` или `insert`, если вам не требуется общее состояние сеанса.
3. Отключите сеансы для общего клиента, установив `autogenerate_session_id=False` перед созданием клиента (или передайте его напрямую в `get_client`).
```python theme={null}
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`:
```python theme={null}
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`](https://urllib3.readthedocs.io/en/stable/advanced-usage.html#customizing-pool-behavior).
Чтобы задать параметры сокета, передайте `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)`.