Arrow Flight インターフェイス

ClickHouse は Apache Arrow Flight プロトコルをサポートしています。これは、gRPC 上で Arrow IPC フォーマットを使用し、効率的な列指向データ転送を実現する高性能な RPC フレームワークです。

この実装には Arrow Flight SQL のサポートも含まれており、Flight SQL プロトコルに対応した BI ツールやアプリケーションから ClickHouse に直接クエリできます。

主な機能:

• SQL クエリを実行し、結果を Apache Arrow フォーマットで取得できます。
• Arrow フォーマットを使用してテーブルにデータを挿入できます。
• Flight SQL コマンドを使ってメタデータ (カタログ、スキーマ、テーブル、主キー) をクエリできます。
• Flight SQL を介してサーバー側のプリペアドステートメントを作成、バインド、実行、クローズできます。
• Flight SQL アクションを介してセッションと設定を管理できます。
• TLS 暗号化とユーザー名/パスワード認証。
• PollFlightInfo による段階的な結果取得。
• CancelFlightInfo によるクエリのキャンセル。

Arrow Flight Server を有効にするには、ClickHouse サーバー設定に arrowflight_port 設定を追加します。

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

起動時に、インターフェイスが有効になっていることを示すログメッセージが表示されます。

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

Arrow Flight インターフェイスで TLS を有効にするには、以下の設定を行います。

<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>

TLS が有効な場合、クライアントは grpc:// ではなく grpc+tls:// 認証スキームを使用して接続する必要があります。

Arrow Flight インターフェイスでは、2 つの認証方法がサポートされています。

クライアントは、標準の HTTP Authorization: Basic ヘッダーを介して、ユーザー名とパスワードで認証します。認証に成功すると、サーバーはレスポンスヘッダーで Bearer トークンを返します。

後続のリクエストでは、基本認証で返されたベアラートークンを Authorization: Bearer <token> ヘッダーで使用できます。トークンは使用するたびに自動的に更新され、有効期限は default_session_timeout サーバー設定 (既定値: 60 秒) に従います。

import pyarrow.flight as flight

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

# 基本認証は後続の呼び出しに使用するベアラートークンを返す
token_pair = client.authenticate_basic_token("default", "")
options = flight.FlightCallOptions(headers=[token_pair])

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])

Arrow Flight インターフェイスは、カスタム gRPC メタデータヘッダーを介して ClickHouse のセッションをサポートします。

セッションを使用すると、SetSessionOptions アクションで永続的な ClickHouse 設定を指定できます (DoAction を参照) 。

クエリを実行し、結果のスキーマ、データ取得用チケット付きのエンドポイント、行数、バイト数を含む FlightInfo を返します。

受け取る FlightDescriptor には、次のいずれかを指定できます。

• PATH descriptor: テーブル名として解釈される単一要素の path です。SELECT * FROM <table> を生成します。
• CMD descriptor: 生の SQL クエリ文字列、またはシリアライズされた Flight SQL protobuf コマンドです (Flight SQL Commands を参照) 。

クエリは最後まで実行され、結果はサーバー側のチケットに保存されます。データの各 block ごとに個別のエンドポイント/チケットが生成されるため、クライアントはデータを並列に取得できます。

# テーブル名でクエリ
descriptor = flight.FlightDescriptor.for_path("my_table")
info = client.get_flight_info(descriptor, options)

# SQLでクエリ
descriptor = flight.FlightDescriptor.for_command(
    "SELECT * FROM my_table WHERE id > 100"
)
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())

長時間実行されるクエリの結果を、段階的に取得できるようにします。GetFlightInfo のようにクエリ全体の完了を待つのではなく、PollFlightInfo は結果をブロック単位で返します。

最初の呼び出しでクエリの実行が開始され、レスポンスには次の内容が含まれます。

• その時点で利用可能なデータブロックに対応するエンドポイントを含む FlightInfo
• 次回のポーリングに使用する FlightDescriptor (さらに結果が返される見込みがある場合)

返されたディスクリプタを使った後続の呼び出しでは、追加のブロックを取得できます。これ以上利用可能なデータがない場合、レスポンスには次のディスクリプタは含まれません。

クエリ全体を実行せずに、クエリ結果の Arrow スキーマを返します。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

指定された ticket に対応するデータを取得します。次のいずれかを受け付けます。

• GetFlightInfo または PollFlightInfo が返す ticket。
• ticket の値として指定する、生の SQL query 文字列。

# GetFlightInfoから取得したticketを使用する場合
reader = client.do_get(endpoint.ticket, options)
table = reader.read_all()

# ticketとして生のSQLクエリを使用する場合
ticket = flight.Ticket("SELECT number FROM system.numbers LIMIT 10")
reader = client.do_get(ticket, options)
table = reader.read_all()

ClickHouse にデータを送信します。FlightDescriptor と Arrow レコードバッチのストリームを受け取ります。

テーブル名による insert (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()

SQLによる挿入 (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()

Flight SQL CommandStatementUpdate による DDL/DML の実行:

Flight SQL クライアントは、DDL/DML ステートメント (CREATE、INSERT、ALTER など) の実行に CommandStatementUpdate を使用します。レスポンスには、影響を受けた行数が含まれます。

Flight SQL CommandStatementIngest による一括取り込み:

サポートされているのは、既存のテーブルへの追記のみです (TABLE_NOT_EXIST_OPTION_FAIL + TABLE_EXISTS_OPTION_APPEND) 。このコマンドでは、カタログと一時テーブルはサポートされていません。

transaction_id は CommandStatementUpdate と CommandStatementIngest ではサポートされていません。指定した場合、ClickHouse は NotImplemented エラーを返します。

名前付きアクションを実行します。サポートされているアクションは次のとおりです。

FlightInfo に関連付けられた実行中のクエリをキャンセルします。クエリ ID は FlightInfo の app_metadata フィールドから抽出されます。また、そのクエリに関連付けられたすべての poll ディスクリプタもキャンセルします。

# PollFlightInfo経由で長時間実行クエリを開始し、キャンセルする
cancel_request = flight.CancelFlightInfoRequest(info)
result = client.cancel_flight_info(cancel_request, options)
# 成功した場合、result.status は CancelStatus.CANCELLED になる

現在のセッションに対する ClickHouse のサーバー設定を行います。x-clickhouse-session-id ヘッダーでセッション ID が設定されている必要があります。

サポートされる値の型: string、boolean、integer、double、および string list。

設定名が不明な場合は、error INVALID_NAME が返されます。値をパースできない場合は、error INVALID_VALUE が返されます。

現在のセッションの ClickHouse 設定とその値をすべて返します。設定名から文字列値へのマップを返します (内部的には system.settings にクエリします) 。

サーバー側のプリペアドステートメントを作成し、ステートメントハンドルを返します。リクエストには、? プレースホルダーを含む SQL クエリテキストが含まれます。

このアクションでは transaction_id はサポートされていません。指定した場合、ClickHouse は NotImplemented エラーを返します。

クエリステートメントの場合、レスポンスには次が含まれることがあります。

• dataset_schema: 結果セットのスキーマ。
• parameter_schema: ステートメントパラメータのスキーマ。

有効なクエリでスキーマ推論に失敗した場合でも (たとえば、そのクエリではプレースホルダーを NULL に置き換えることが有効でない場合) 、ClickHouse はプリペアドステートメントを作成し、dataset_schema を含めずにハンドルを返します。

プリペアドステートメントは、単一のセッションではなく、認証されたユーザーに属します。同じユーザーとして複数のセッションを開いている場合、それらのどのセッションからでも同じステートメントハンドルを実行、再バインド、クローズできます。

他のユーザーは、自分で作成していないステートメントハンドルを実行、バインド、またはクローズできません。

arrowflight.prepared_statements_lifetime_seconds は有効期限の動作を制御します。

• > 0: 設定された値をステートメントの有効期間として使用します。セッションに紐づくステートメントとセッションに紐づかないステートメントの両方で、リクエストのたびに有効期限が更新されます。
• 0: プリペアドステートメントは自動的に期限切れになりません。
• -1 (デフォルト): ステートメントがセッション内で作成された場合、その有効期間はそのセッションのタイムアウトに従い、そのセッション内のリクエストごとに更新されます。セッションなしでステートメントが作成された場合、自動的に期限切れにはなりません。

期限切れになったステートメントは削除され、arrowflight.max_prepared_statements_per_user のカウント対象にも含まれなくなります。

リクエストに空でないステートメントハンドルが含まれている場合、プリペアドステートメントを閉じ、関連するサーバー側リソースを解放します。

ClickHouse は、ハンドルが空の場合、ClosePreparedStatement による一括クローズもサポートしています。

• x-clickhouse-session-id が存在する場合、そのセッション内で認証済みユーザーのすべてのプリペアドステートメントを閉じます。
• セッション ID が存在しない場合、認証済みユーザーのセッションに属さないプリペアドステートメントのみを閉じます。

プリペアドステートメントがセッション内で (x-clickhouse-session-id を介して) 作成された場合、そのセッションが閉じられると、そのステートメントも自動的に閉じられます。

CMD ディスクリプタにシリアライズされた Flight SQL protobuf メッセージが含まれている場合、ClickHouse は以下のコマンドを処理します。

これらのコマンドは ClickHouse が提供していない機能に対応しているため、Arrow Flight SQL インターフェイスではサポートされていません。

import pyarrow as pa
import pyarrow.flight as flight

# 接続と認証
client = flight.FlightClient("grpc://localhost:9090")
token = client.authenticate_basic_token("default", "")
options = flight.FlightCallOptions(headers=[token])

# PATH ディスクリプタを使用した DoPut によるデータ挿入
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()

# 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

すべてのデータは Apache Arrow IPC フォーマットで転送されます。サポートされるのは Arrow フォーマットのみで、ほかの ClickHouse フォーマット (例: FORMAT JSON、FORMAT CSV) を指定するとエラーになります。

ClickHouse のデータ型は、シリアライゼーション時に Arrow 型にマッピングされます。設定 output_format_arrow_unsupported_types_as_binary は、未サポートの ClickHouse データ型をバイナリブロブとしてシリアライズするかどうかを制御します。

Arrow Flight インターフェイスは、Arrow Flight または Arrow Flight SQL プロトコルをサポートするあらゆるクライアントやツールと互換性があります。具体的には、次のようなものが含まれます。

• Python (pyarrow)
• Java (org.apache.arrow.flight)
• C++ (arrow::flight)
• Go (apache/arrow/go)
• ADBC (Arrow Database Connectivity) ドライバー
• DBeaver など、Flight SQL をサポートするその他のツール

お使いのツール向けにネイティブの ClickHouse コネクタ (例: JDBC、ODBC、ネイティブプロトコル) が利用できる場合は、パフォーマンス上の理由やフォーマット互換性のために Arrow Flight が特に必要でない限り、そちらを優先して使用してください。

ClickHouseは、外部のArrow Flightサーバーからデータを読み取るFlightクライアントとしても動作します。詳しくは、以下を参照してください。

• ArrowFlightテーブルエンジン
• arrowFlightテーブル関数

• Apache Arrow Flight 仕様
• Apache Arrow Flight SQL 仕様
• ClickHouse の Arrow フォーマット