JDBC コネクタ

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

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

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

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

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

テーブルが作成されていることを確認し、以前の例ですでに存在している場合は削除してください。縮小版の 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 を スタンドアロン モードまたは 分散 モードで起動します。

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

提供されているスクリプトと設定を使用して、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 |