JDBC コネクタ

Kafka Connect と ClickHouse で JDBC Connector Sink を使用する

​

JDBC コネクタ

このコネクタは、データがシンプルで、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 管理者によって設定されます。

​

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

Confluent パッケージをダウンロードし、ローカル環境にインストール済みであることを前提としています。コネクタのインストールについては、こちらに記載されている手順に従ってください。

confluent-hub を使ったインストール方法を使用する場合は、ローカルの設定ファイルが更新されます。

Kafka から ClickHouse にデータを送信するには、コネクタの sink コンポーネントを使用します。

​

2. 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 を変更してください。詳細は以下を参照してください。

​

3. 設定を準備する

インストール形態に応じた 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 でホストしていることを前提としています。

​

4. ClickHouseテーブルを作成する

テーブルが作成されていることを確認し、前の例ですでに存在する場合は削除してください。縮小版のGitHubデータセットに対応した例を以下に示します。現時点ではサポートされていない 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)

​

5. Kafka Connect を起動する

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

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

​

6. Kafka にデータを追加する

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

python producer.py -c github.config

このスクリプトを使うと、任意の ndjson ファイルを Kafka トピックに挿入できます。スキーマは自動的に推論されます。付属のサンプル設定で挿入されるのは 10k メッセージのみです。必要に応じて こちらで変更 してください。この設定では、Kafka への挿入時に、互換性のない Array フィールドもデータセットから削除されます。

これは、JDBC コネクタがメッセージを INSERT ステートメントに変換するために必要です。独自のデータを使用する場合は、各メッセージにスキーマを含めて挿入する (_value.converter.schemas.enable _を true に設定する) か、クライアントがレジストリ内のスキーマを参照するメッセージをパブリッシュするようにしてください。

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 詳説 – コンバータとシリアライゼーションの解説