dlt を ClickHouse に接続する

dlt は、さまざまな、しかもしばしば雑然としたデータソースから、適切に構造化されたライブデータセットへデータを読み込むために、Python スクリプトに追加できるオープンソースのライブラリです。

pip install "dlt[clickhouse]"

dlt init chess clickhouse

pip install -r requirements.txt

CREATE DATABASE IF NOT EXISTS dlt;
CREATE USER dlt IDENTIFIED WITH sha256_password BY 'Dlt*12345789234567';
GRANT CREATE, ALTER, SELECT, DELETE, DROP, TRUNCATE, OPTIMIZE, SHOW, INSERT, dictGet ON dlt.* TO dlt;
GRANT SELECT ON INFORMATION_SCHEMA.COLUMNS TO dlt;
GRANT CREATE TEMPORARY TABLE, S3 ON *.* TO dlt;

[destination.clickhouse.credentials]
database = "dlt"                         # 作成したデータベース名
username = "dlt"                         # ClickHouse のユーザー名。通常のデフォルトは "default"
password = "Dlt*12345789234567"          # 必要に応じて ClickHouse のパスワード
host = "localhost"                       # ClickHouse サーバーのホスト
port = 9000                              # ClickHouse の HTTP ポート。デフォルトは 9000
http_port = 8443                         # ClickHouse サーバーの HTTP インターフェイスに接続するための HTTP ポート。デフォルトは 8443。
secure = 1                               # HTTPS を使用する場合は 1、それ以外は 0 に設定します。

[destination.clickhouse]
dataset_table_separator = "___"          # データセット から生成される データセット テーブル名の区切り文字。

# toml ファイルの先頭、どのセクションよりも前に置いてください。
destination.clickhouse.credentials="clickhouse://dlt:Dlt*12345789234567@localhost:9000/dlt?secure=1"

すべての書き込みディスポジション に対応しています。

dlt ライブラリの書き込みディスポジションは、データを宛先にどのように書き込むかを定義するものです。書き込みディスポジションには、次の 3 種類があります。

Replace: このディスポジションでは、宛先のデータをリソースのデータで置き換えます。データをロードする前に、すべてのクラスとオブジェクトを削除し、スキーマを再作成します。詳しくはこちらをご覧ください。

Merge: この書き込みディスポジションでは、リソースのデータを宛先のデータにマージします。merge ディスポジションを使用する場合は、リソースに primary_key を指定する必要があります。詳しくはこちらをご覧ください。

Append: これはデフォルトのディスポジションです。primary_key フィールドは無視され、データは宛先内の既存データに追記されます。

データは、データソースに応じて最も効率的な方法で ClickHouse に読み込まれます。

• ローカルファイルの場合は、clickhouse-connect ライブラリを使用して、INSERT コマンドでファイルを ClickHouse テーブルに直接読み込みます。
• S3, Google Cloud Storage, または Azure Blob Storage などのリモートストレージ上のファイルの場合は、s3、gcs、azureBlobStorage などの ClickHouse テーブル関数を使用してファイルを読み込み、データをテーブルに挿入します。

ClickHouse は 1 つのデータベース内で複数のデータセットをサポートしていませんが、dlt はいくつかの理由からデータセットを前提としています。ClickHouse を dlt で利用できるようにするため、ClickHouse データベース内で dlt によって生成されるテーブル名には、設定可能な dataset_table_separator で区切られたデータセット名のプレフィックスが付きます。さらに、データを一切含まない特別なセンチネルテーブルが作成され、dlt はこれによって ClickHouse の宛先にどの仮想データセットがすでに存在するかを認識できます。

• jsonl は、直接ロードとステージングの両方で推奨されるフォーマットです。
• parquet は、直接ロードとステージングの両方でサポートされています。

clickhouse 宛先には、デフォルトの SQL 宛先とはいくつか異なる点があります。

1. ClickHouse には実験的な object データ型がありますが、やや予測不能な挙動をすることがあるため、dlt の clickhouse 宛先では複雑なデータ型はテキストカラムにロードされます。この機能が必要な場合は、Slack コミュニティでご相談ください。追加を検討します。
2. ClickHouse は time データ型をサポートしていません。time は text カラムにロードされます。
3. ClickHouse は binary データ型をサポートしていません。代わりに、バイナリデータは text カラムにロードされます。jsonl からロードする場合、バイナリデータは base64 文字列になり、parquet からロードする場合は binary オブジェクトが text に変換されます。
4. ClickHouse では、データがすでに入っているテーブルに対しても、NULL 不可のカラムを追加できます。
5. ClickHouse は、float または double データ型を使用すると、特定の条件下で丸め誤差が生じることがあります。丸め誤差を許容できない場合は、decimal データ型を使用してください。たとえば、ローダーのファイルフォーマットを jsonl に設定して値 12.7001 を double カラムにロードすると、確実に丸め誤差が発生します。

ClickHouse は、以下のカラムヒントをサポートしています。

• primary_key - このカラムを主キーの一部としてマークします。複数のカラムにこのヒントを設定することで、複合主キーを作成できます。

デフォルトでは、ClickHouse ではテーブルは ReplicatedMergeTree テーブルエンジンで作成されます。clickhouse アダプターでは、table_engine_type を使用して別のテーブルエンジンを指定できます。

from dlt.destinations.adapters import clickhouse_adapter

@dlt.resource()
def my_resource():
  ...

clickhouse_adapter(my_resource, table_engine_type="merge_tree")

サポートされている値は次のとおりです。

• merge_tree - MergeTree エンジンを使用してテーブルを作成します
• replicated_merge_tree (デフォルト) - ReplicatedMergeTree エンジンを使用してテーブルを作成します

ClickHouse は、ファイルのステージング先として Amazon S3、Google Cloud Storage、Azure Blob Storage をサポートしています。

dlt は Parquet または jsonl ファイルをステージング先にアップロードし、ClickHouse のテーブル関数を使用して、ステージングされたファイルからデータを直接読み込みます。

ステージング先の認証情報を設定する方法については、filesystem のドキュメントを参照してください。

• Amazon S3
• Google Cloud Storage
• Azure Blob Storage

ステージングを有効にしてパイプラインを実行するには:

pipeline = dlt.pipeline(
  pipeline_name='chess_pipeline',
  destination='clickhouse',
  staging='filesystem',  # ステージングを有効にするために追加
  dataset_name='chess_data'
)

dlt は、ClickHouse にデータを読み込む際のステージング領域として Google Cloud Storage (GCS) を使用することをサポートしています。これは、dlt が内部的に使用する ClickHouse の GCS テーブル関数 によって自動的に処理されます。

ClickHouse の GCS テーブル関数 は、Hash-based Message Authentication Code (HMAC) キーを使用した認証のみをサポートしています。これを有効にするため、GCS は Amazon S3 API をエミュレートする S3 互換性モードを提供しています。ClickHouse はこれを利用して、S3 インテグレーション経由で GCS バケットにアクセスできるようにしています。

dlt で HMAC 認証を使用した GCS ステージングを設定するには:

1. Google Cloud ガイド に従って、GCS のサービス アカウント用の HMAC キーを作成します。

2. dlt プロジェクトの config.toml にある ClickHouse の宛先設定で、HMAC キーに加えて、サービス アカウントの client_email、project_id、private_key を設定します:

[destination.filesystem]
bucket_url = "gs://dlt-ci"

[destination.filesystem.credentials]
project_id = "a-cool-project"
client_email = "my-service-account@a-cool-project.iam.gserviceaccount.com"
private_key = "-----BEGIN PRIVATE KEY-----\nMIIEvQIBADANBgkaslkdjflasjnkdcopauihj...wEiEx7y+mx\nNffxQBqVVej2n/D93xY99pM=\n-----END PRIVATE KEY-----\n"

[destination.clickhouse.credentials]
database = "dlt"
username = "dlt"
password = "Dlt*12345789234567"
host = "localhost"
port = 9440
secure = 1
gcp_access_key_id = "JFJ$$*f2058024835jFffsadf"
gcp_secret_access_key = "DFJdwslf2hf57)%$02jaflsedjfasoi"

注: HMAC キー bashgcp_access_key_id と gcp_secret_access_key) に加えて、[destination.filesystem.credentials] の下で、サービスアカウントの client_email、project_id、private_key も指定する必要があります。これは、GCS のステージング サポートが現在は一時的な回避策として実装されており、まだ最適化されていないためです。

dlt はこれらの認証情報を ClickHouse に渡し、ClickHouse が認証と GCS へのアクセスを処理します。

今後、ClickHouse の dlt 宛先向け GCS ステージング設定を簡素化し、改善するための作業が活発に進められています。正式な GCS ステージング サポートは、以下の GitHub issue で追跡されています。

• ファイルシステム宛先を gcs の s3 互換性モードで動作させる
• Google Cloud Storage ステージング エリアのサポート

dbt とのインテグレーションは、通常 dbt-clickhouse を通じてサポートされています。

この宛先は、dlt の状態同期に完全対応しています。