JDBC コネクタ - ClickHouse Documentation

このコネクタは、データがシンプルで、int などのプリミティブなデータ型だけで構成されている場合にのみ使用してください。Map などの ClickHouse 固有の型はサポートされていません。

この例では、Confluent の Kafka Connect ディストリビューションを使用します。以下では、単一の Kafka トピックからメッセージを取得し、ClickHouse テーブルに行を挿入するシンプルなインストール方法を説明します。Kafka 環境がない場合は、無料 tier が充実している Confluent Cloud を推奨します。 JDBC コネクタではスキーマが必須である点に注意してください (JDBC コネクタではプレーンな JSON や CSV は使用できません) 。スキーマは各メッセージにエンコードすることもできますが、その場合のオーバーヘッドを避けるため、Confluent スキーマレジストリの使用を強く推奨します。提供されている挿入スクリプトは、メッセージから自動的にスキーマを推論してレジストリに登録するため、他のデータセットにも再利用できます。Kafka のキーは String であることを前提としています。Kafka のスキーマの詳細は、こちらを参照してください。

ライセンス

JDBC コネクタは、Confluent Community License に基づいて配布されています

手順

接続情報を確認する

HTTP(S) で ClickHouse に接続するには、次の情報が必要です。

Parameter(s)	Description
`HOST` and `PORT`	通常、TLS を使用する場合のポートは 8443、TLS を使用しない場合は 8123 です。
`DATABASE NAME`	デフォルトでは `default` という名前のデータベースがあります。接続先のデータベース名を使用してください。
`USERNAME` and `PASSWORD`	デフォルトのユーザー名は `default` です。用途に応じたユーザー名を使用してください。

ClickHouse Cloud サービスの詳細は、ClickHouse Cloud コンソールで確認できます。サービスを選択し、Connect をクリックします。

HTTPS を選択します。接続情報は curl コマンドの例として表示されます。

セルフマネージド ClickHouse を使用している場合、接続情報は ClickHouse 管理者によって設定されます。

Kafka Connect とコネクタをインストールする

Confluent パッケージをダウンロードし、ローカル環境にインストール済みであることを前提としています。コネクタのインストールについては、こちらに記載されている手順に従ってください。confluent-hub を使用してインストールする場合、ローカルの設定ファイルが更新されます。Kafka から ClickHouse にデータを送信するには、コネクタの Sink コンポーネントを使用します。

JDBC ドライバーをダウンロードしてインストールする

こちらから ClickHouse JDBCドライバー clickhouse-jdbc-<version>-shaded.jar をダウンロードしてインストールしてください。Kafka Connect へのインストールは、こちらの手順に従ってください。他のドライバーでも動作する可能性はありますが、テストは行っていません。

よくある問題: ドキュメントでは jar を share/java/kafka-connect-jdbc/ にコピーするよう案内されています。Connect がドライバーを見つけられない場合は、ドライバーを share/confluent-hub-components/confluentinc-kafka-connect-jdbc/lib/ にコピーしてください。あるいは、plugin.path を変更してドライバーを含めてください。詳細は以下を参照してください。

設定を準備する

インストール形態に応じた Connect の設定については、スタンドアロンクラスターと分散クラスターの違いに注意しつつ、こちらの手順に従ってください。Confluent Cloud を使用する場合は、分散構成が該当します。以下のパラメータは、ClickHouse で JDBC コネクタを使用する際に重要です。パラメータの完全な一覧はこちらで確認できます。

_connection.url_ - jdbc:clickhouse://<clickhouse host>:<clickhouse http port>/<target database> の形式で指定する必要があります
connection.user - 移行先データベースへの書き込み権限を持つユーザー
table.name.format- データの挿入先となる ClickHouse テーブル。事前に存在している必要があります。
batch.size - 1 回のバッチで送信する行数です。適切に大きな値を設定してください。ClickHouse の推奨事項では、1000 を最低値の目安とすることを推奨しています。
tasks.max - JDBC Sink コネクタは 1 つ以上の task の実行をサポートしています。これはパフォーマンス向上に利用できます。batch size とあわせて、パフォーマンス改善の主要な手段となります。
value.converter.schemas.enable - スキーマレジストリを使用する場合は false、メッセージ内にスキーマを埋め込む場合は true に設定します。
value.converter - データ型に応じて設定します。たとえば JSON の場合は io.confluent.connect.json.JsonSchemaConverter です。
key.converter - org.apache.kafka.connect.storage.StringConverter に設定します。String の key を利用します。
pk.mode - ClickHouse では関係ありません。none に設定します。
auto.create - サポートされていないため、false にする必要があります。
auto.evolve - 将来的にサポートされる可能性はありますが、この設定は false を推奨します。
insert.mode - “insert” に設定します。現在、他のモードはサポートされていません。
key.converter - key の型に応じて設定します。
value.converter - topic 上のデータ型に応じて設定します。このデータには、JSON、Avro、または Protobuf フォーマットのいずれかでサポートされるスキーマが必要です。

テストにサンプルデータセットを使用する場合は、以下が設定されていることを確認してください。

value.converter.schemas.enable - スキーマレジストリを利用するため false に設定します。各メッセージにスキーマを埋め込む場合は true に設定します。
key.converter - “org.apache.kafka.connect.storage.StringConverter” に設定します。String の key を利用します。
value.converter - “io.confluent.connect.json.JsonSchemaConverter” に設定します。
value.converter.schema.registry.url - スキーマサーバーの URL を設定し、あわせてパラメータ value.converter.schema.registry.basic.auth.user.info でスキーマサーバーの認証情報を設定します。

GitHub のサンプルデータ用の設定ファイル例はこちらにあります。これは、Connect をスタンドアロンモードで実行し、Kafka を Confluent Cloud でホストしていることを前提としています。

ClickHouse テーブルを作成する

テーブルが作成されていることを確認し、以前の例ですでに存在している場合は削除してください。縮小版の GitHub dataset に対応した例を以下に示します。現在サポートされていない Array 型や Map 型が含まれていない点に注意してください:

CREATE TABLE github
(
    file_time DateTime,
    event_type Enum('CommitCommentEvent' = 1, 'CreateEvent' = 2, 'DeleteEvent' = 3, 'ForkEvent' = 4, 'GollumEvent' = 5, 'IssueCommentEvent' = 6, 'IssuesEvent' = 7, 'MemberEvent' = 8, 'PublicEvent' = 9, 'PullRequestEvent' = 10, 'PullRequestReviewCommentEvent' = 11, 'PushEvent' = 12, 'ReleaseEvent' = 13, 'SponsorshipEvent' = 14, 'WatchEvent' = 15, 'GistEvent' = 16, 'FollowEvent' = 17, 'DownloadEvent' = 18, 'PullRequestReviewEvent' = 19, 'ForkApplyEvent' = 20, 'Event' = 21, 'TeamAddEvent' = 22),
    actor_login LowCardinality(String),
    repo_name LowCardinality(String),
    created_at DateTime,
    updated_at DateTime,
    action Enum('none' = 0, 'created' = 1, 'added' = 2, 'edited' = 3, 'deleted' = 4, 'opened' = 5, 'closed' = 6, 'reopened' = 7, 'assigned' = 8, 'unassigned' = 9, 'labeled' = 10, 'unlabeled' = 11, 'review_requested' = 12, 'review_request_removed' = 13, 'synchronize' = 14, 'started' = 15, 'published' = 16, 'update' = 17, 'create' = 18, 'fork' = 19, 'merged' = 20),
    comment_id UInt64,
    path String,
    ref LowCardinality(String),
    ref_type Enum('none' = 0, 'branch' = 1, 'tag' = 2, 'repository' = 3, 'unknown' = 4),
    creator_user_login LowCardinality(String),
    number UInt32,
    title String,
    state Enum('none' = 0, 'open' = 1, 'closed' = 2),
    assignee LowCardinality(String),
    closed_at DateTime,
    merged_at DateTime,
    merge_commit_sha String,
    merged_by LowCardinality(String),
    review_comments UInt32,
    member_login LowCardinality(String)
) ENGINE = MergeTree ORDER BY (event_type, repo_name, created_at)

Kafka Connect を起動する

Kafka Connect をスタンドアロンモードまたは分散モードで起動します。

./bin/connect-standalone connect.properties.ini github-jdbc-sink.properties.ini

Kafka にデータを投入する

提供されているスクリプトと設定を使用して、Kafka にメッセージを送信します。github.config を変更して、Kafka の認証情報を含める必要があります。このスクリプトは現在、Confluent Cloud で使用するように設定されています。

python producer.py -c github.config

このスクリプトを使うと、任意の ndjson ファイルを Kafka topic に挿入できます。スキーマの自動推論も試みます。提供されているサンプル config では 10k 件のメッセージしか挿入されません。必要に応じてこちらで変更してください。この設定では、Kafka への挿入時にデータセットから互換性のない Array フィールドも削除されます。これは、JDBC コネクタがメッセージを INSERT 文に変換するために必要です。独自のデータを使用する場合は、すべてのメッセージにスキーマを含めて挿入する (_value.converter.schemas.enable _を true に設定する) か、client がスキーマレジストリ内のスキーマを参照するメッセージを公開するようにしてください。Kafka Connect はメッセージの消費を開始し、ClickHouse への行の挿入を開始するはずです。“[JDBC Compliant Mode] Transaction isn’t supported.” に関する警告は想定内であり、無視してかまいません。ターゲットテーブル “Github” を単純に読み取れば、データが挿入されたことを確認できるはずです。

SELECT count() FROM default.github;

| count\(\) |
| :--- |
| 10000 |

参考情報

Kafka Sink の設定パラメータ
Kafka Connect 詳説 – JDBC ソースコネクタ
Kafka Connect JDBC Sink 詳説: 主キーの扱い
Kafka Connect 実践: JDBC Sink - 読むより視聴したい方はこちら。
Kafka Connect 詳説 – コンバータとシリアライゼーションの解説

​ライセンス

​手順

​接続情報を確認する

​参考情報

ライセンス

手順

接続情報を確認する

参考情報