Interfaz Arrow Flight

ClickHouse admite el protocolo Apache Arrow Flight: un framework de RPC de alto rendimiento para el transporte eficiente de datos en formato columnar mediante Arrow IPC sobre gRPC.

La implementación incluye compatibilidad con Arrow Flight SQL, lo que permite a las herramientas de BI y a las aplicaciones que usan el protocolo Flight SQL consultar ClickHouse directamente.

Capacidades clave:

• Ejecutar consultas SQL y recuperar resultados en formato Apache Arrow.
• Insertar datos en tablas con el formato Arrow.
• Consultar metadatos (catálogos, esquemas, tablas y claves primarias) mediante comandos de Flight SQL.
• Crear, vincular, ejecutar y cerrar sentencias preparadas en el servidor mediante Flight SQL.
• Administrar sesiones y ajustes mediante acciones de Flight SQL.
• Cifrado TLS y autenticación mediante nombre de usuario y contraseña.
• Recuperación incremental de resultados mediante PollFlightInfo.
• Cancelación de consultas mediante CancelFlightInfo.

Para habilitar el servidor Arrow Flight, añada el ajuste arrowflight_port a la configuración del servidor de ClickHouse:

<clickhouse>
    <arrowflight_port>9090</arrowflight_port>
</clickhouse>

Al iniciarse, un mensaje en el registro confirma que la interfaz está activa:

{} <Information> Application: Arrow Flight compatibility protocol: 0.0.0.0:9090

Para habilitar TLS en la interfaz Arrow Flight, configure los siguientes ajustes:

<clickhouse>
    <arrowflight_port>9090</arrowflight_port>
    <arrowflight>
        <enable_ssl>true</enable_ssl>
        <ssl_cert_file>/path/to/server-cert.pem</ssl_cert_file>
        <ssl_key_file>/path/to/server-key.pem</ssl_key_file>
    </arrowflight>
</clickhouse>

Cuando TLS está habilitado, los clientes deben conectarse con el esquema grpc+tls:// en lugar de grpc://.

La interfaz Arrow Flight admite dos métodos de autenticación:

Los clientes se autentican con un nombre de usuario y una contraseña mediante el encabezado HTTP estándar Authorization: Basic. Tras autenticarse correctamente, el servidor devuelve un token Bearer en el encabezado de la respuesta.

Las solicitudes posteriores pueden usar el token Bearer devuelto por la autenticación básica a través del encabezado Authorization: Bearer <token>. El token se renueva automáticamente con cada uso y caduca según la configuración del servidor default_session_timeout (valor predeterminado: 60 segundos).

import pyarrow.flight as flight

client = flight.FlightClient("grpc://localhost:9090")

# La autenticación básica devuelve un token Bearer para llamadas posteriores
token_pair = client.authenticate_basic_token("default", "")
options = flight.FlightCallOptions(headers=[token_pair])

Con TLS:

import pyarrow.flight as flight

with open("ca-cert.pem", "rb") as f:
    tls_root_certs = f.read()

client = flight.FlightClient(
    "grpc+tls://localhost:9090",
    tls_root_certs=tls_root_certs,
)

token_pair = client.authenticate_basic_token("default", "password")
options = flight.FlightCallOptions(headers=[token_pair])

La interfaz Arrow Flight admite sesiones de ClickHouse mediante encabezados de metadatos gRPC personalizados:

Las sesiones permiten establecer ajustes persistentes de ClickHouse mediante la acción SetSessionOptions (consulte DoAction).

Ejecuta una consulta y devuelve un FlightInfo que contiene el esquema del resultado, endpoints con tickets para recuperar los datos, el número de filas y la cantidad de bytes.

Acepta un FlightDescriptor, que puede ser:

• PATH descriptor: Una ruta de un solo componente interpretada como nombre de tabla. Genera SELECT * FROM <table>.
• CMD descriptor: Una cadena de consulta SQL sin procesar o un comando protobuf serializado de Flight SQL (consulta Comandos de Flight SQL).

La consulta se ejecuta por completo y los resultados se almacenan en tickets en el servidor. Cada bloque de datos genera un endpoint/ticket independiente, lo que permite a los clientes recuperar los datos en paralelo.

# Consulta por nombre de tabla
descriptor = flight.FlightDescriptor.for_path("my_table")
info = client.get_flight_info(descriptor, options)

# Consulta por SQL
descriptor = flight.FlightDescriptor.for_command(
    "SELECT * FROM my_table WHERE id > 100"
)
info = client.get_flight_info(descriptor, options)

# Recuperar resultados
for endpoint in info.endpoints:
    reader = client.do_get(endpoint.ticket, options)
    table = reader.read_all()
    print(table.to_pandas())

Permite recuperar resultados de forma incremental en consultas de larga duración. En lugar de esperar a que se complete toda la consulta (como hace GetFlightInfo), PollFlightInfo devuelve los resultados bloque por bloque.

En la primera llamada, la consulta empieza a ejecutarse. La respuesta incluye:

• Un FlightInfo con endpoints para todos los bloques de datos disponibles hasta ese momento.
• Un FlightDescriptor para el siguiente sondeo (si se esperan más resultados).

Las llamadas posteriores con el descriptor devuelto recuperan bloques adicionales. Cuando ya no hay más datos disponibles, la respuesta no incluye un descriptor para el siguiente sondeo.

Devuelve el esquema de Arrow del resultado de una consulta sin ejecutar la consulta completa. Acepta los mismos tipos de descriptor que GetFlightInfo.

descriptor = flight.FlightDescriptor.for_command(
    "SELECT 1 AS x, 'hello' AS y"
)
schema_result = client.get_schema(descriptor, options)
schema = schema_result.schema
print(schema)  # x: int32, y: string

Recupera datos para un ticket determinado. Acepta una de estas opciones:

• Un ticket devuelto por GetFlightInfo o PollFlightInfo.
• Una cadena con una consulta SQL sin procesar como valor del ticket.

# Usando un ticket de GetFlightInfo
reader = client.do_get(endpoint.ticket, options)
table = reader.read_all()

# Usando una consulta SQL sin procesar como ticket
ticket = flight.Ticket("SELECT number FROM system.numbers LIMIT 10")
reader = client.do_get(ticket, options)
table = reader.read_all()

Envía datos a ClickHouse. Acepta un FlightDescriptor y un flujo de lotes de registros de Arrow.

Inserción por nombre de tabla (descriptor PATH):

schema = pa.schema([("id", pa.int64()), ("name", pa.string())])
batch = pa.record_batch(
    [pa.array([1, 2, 3]), pa.array(["Alice", "Bob", "Charlie"])],
    schema=schema,
)

descriptor = flight.FlightDescriptor.for_path("my_table")
writer, _ = client.do_put(descriptor, schema, options)
writer.write_batch(batch)
writer.close()

Inserción por SQL (descriptor CMD):

descriptor = flight.FlightDescriptor.for_command(
    "INSERT INTO my_table FORMAT Arrow"
)
writer, _ = client.do_put(descriptor, schema, options)
writer.write_batch(batch)
writer.close()

Ejecutar DDL/DML mediante Flight SQL CommandStatementUpdate:

Los clientes de Flight SQL usan CommandStatementUpdate para ejecutar sentencias DDL/DML (CREATE, INSERT, ALTER, etc.). La respuesta incluye el número de filas afectadas.

Ingesta masiva mediante Flight SQL CommandStatementIngest:

Solo se admite agregar datos a tablas existentes (TABLE_NOT_EXIST_OPTION_FAIL + TABLE_EXISTS_OPTION_APPEND). Los catálogos y las tablas temporales no son compatibles con este comando.

transaction_id no es compatible con CommandStatementUpdate ni con CommandStatementIngest. Si se proporciona, ClickHouse devuelve un error NotImplemented.

Ejecuta acciones identificadas por nombre. Se admiten las siguientes acciones:

Cancela una consulta en ejecución asociada a un FlightInfo. El ID de la consulta se extrae del campo app_metadata de FlightInfo. También cancela cualquier descriptor de sondeo asociado a la consulta.

# Inicia una consulta de larga duración mediante PollFlightInfo y luego cancélala
cancel_request = flight.CancelFlightInfoRequest(info)
result = client.cancel_flight_info(cancel_request, options)
# result.status es CancelStatus.CANCELLED si se realizó correctamente

Establece la configuración del servidor de ClickHouse para la sesión actual. Requiere que se haya establecido un ID de sesión mediante el encabezado x-clickhouse-session-id.

Tipos de valores admitidos: string, boolean, integer, double y listas de string.

Si no se reconoce el nombre de una configuración, se devuelve el error INVALID_NAME. Si no se puede interpretar un valor, se devuelve el error INVALID_VALUE.

Devuelve todos los ajustes actuales de ClickHouse y sus valores de la sesión. Devuelve un mapa de nombres de ajustes a valores de tipo cadena (consulta system.settings internamente).

Crea una sentencia preparada en el servidor y devuelve un identificador de sentencia. La solicitud contiene el texto de la consulta SQL con marcadores de posición ?.

transaction_id no es compatible con esta acción. Si se proporciona, ClickHouse devuelve un error NotImplemented.

Para las sentencias de consulta, la respuesta puede incluir:

• dataset_schema: esquema del conjunto de resultados.
• parameter_schema: esquema de los parámetros de la sentencia.

Si la inferencia de esquema falla para una consulta válida (por ejemplo, cuando sustituir los marcadores de posición por NULL no es válido para esa consulta), ClickHouse igualmente crea la sentencia preparada y devuelve el identificador sin dataset_schema.

Las sentencias preparadas pertenecen al usuario autenticado, no a una sola sesión. Si abre varias sesiones como el mismo usuario, puede ejecutar, volver a enlazar y cerrar el mismo identificador de sentencia desde cualquiera de esas sesiones.

Otros usuarios no pueden ejecutar, enlazar ni cerrar un identificador de sentencia que no hayan creado.

arrowflight.prepared_statements_lifetime_seconds controla el comportamiento de expiración:

• > 0: usa el valor configurado como tiempo de vida de la sentencia. La expiración se renueva con cada solicitud, tanto para las sentencias vinculadas a una sesión como para las que no tienen sesión.
• 0: las sentencias preparadas no expiran automáticamente.
• -1 (predeterminado): si la sentencia se crea en una sesión, su tiempo de vida sigue el tiempo de espera de esa sesión y se renueva con cada solicitud de esa sesión. Si la sentencia se crea sin una sesión, no expira automáticamente.

Las sentencias expiradas se eliminan y dejan de contabilizarse para arrowflight.max_prepared_statements_per_user.

Cierra una sentencia preparada y libera los recursos asociados del lado del servidor cuando la solicitud contiene un identificador de sentencia no vacío.

ClickHouse también admite el cierre masivo con ClosePreparedStatement cuando el identificador está vacío:

• Si x-clickhouse-session-id está presente, cierra todas las sentencias preparadas del usuario autenticado en esa sesión.
• Si no hay ningún ID de sesión, cierra solo las sentencias preparadas sin sesión del usuario autenticado.

Si una sentencia preparada se crea dentro de una sesión (mediante x-clickhouse-session-id), también se cierra automáticamente cuando se cierra esa sesión.

Cuando un descriptor CMD contiene un mensaje protobuf de Flight SQL serializado, ClickHouse admite los siguientes comandos:

Estos comandos corresponden a funciones que ClickHouse no ofrece, por lo que no son compatibles con la interfaz Arrow Flight SQL.

import pyarrow as pa
import pyarrow.flight as flight

# Conectar y autenticar
client = flight.FlightClient("grpc://localhost:9090")
token = client.authenticate_basic_token("default", "")
options = flight.FlightCallOptions(headers=[token])

# Insertar datos usando DoPut con un descriptor PATH
schema = pa.schema([("id", pa.uint32()), ("value", pa.string())])
batch = pa.record_batch(
    [pa.array([1, 2, 3], type=pa.uint32()), pa.array(["a", "b", "c"])],
    schema=schema,
)
descriptor = flight.FlightDescriptor.for_path("test")
writer, _ = client.do_put(descriptor, schema, options)
writer.write_batch(batch)
writer.close()

# Consultar datos usando GetFlightInfo + DoGet
descriptor = flight.FlightDescriptor.for_command(
    "SELECT * FROM test ORDER BY id"
)
info = client.get_flight_info(descriptor, options)
for endpoint in info.endpoints:
    reader = client.do_get(endpoint.ticket, options)
    table = reader.read_all()
    print(table.to_pandas())

   id value
0   1     a
1   2     b
2   3     c

Todos los datos se transfieren en el formato Apache Arrow IPC. Solo se admite el formato Arrow; si se especifican otros formatos de ClickHouse (por ejemplo, FORMAT JSON, FORMAT CSV), se produce un error.

Los tipos de datos de ClickHouse se asignan a tipos de Arrow durante la serialización. La configuración output_format_arrow_unsupported_types_as_binary controla si los tipos de ClickHouse no compatibles se serializan como blobs binarios.

La interfaz Arrow Flight es compatible con cualquier cliente o herramienta que admita el protocolo Arrow Flight o Arrow Flight SQL, entre ellos:

• Python (pyarrow)
• Java (org.apache.arrow.flight)
• C++ (arrow::flight)
• Go (apache/arrow/go)
• controladores ADBC (Arrow Database Connectivity)
• DBeaver y otras herramientas compatibles con Flight SQL

Si hay un conector nativo de ClickHouse disponible para su herramienta (p. ej., JDBC, ODBC, protocolo nativo), es preferible usarlo, salvo que Arrow Flight sea necesario específicamente por motivos de rendimiento o de compatibilidad de formato.

ClickHouse también puede actuar como cliente de Flight para leer datos de servidores externos de Arrow Flight. Consulte:

• motor de tabla ArrowFlight
• función de tabla arrowFlight

• Especificación de Apache Arrow Flight
• Especificación de Apache Arrow Flight SQL
• Formato Arrow en ClickHouse