> ## Documentation Index > Fetch the complete documentation index at: https://clickhouse.com/docs/llms.txt > Use this file to discover all available pages before exploring further. > Kafka Connect と ClickHouse で JDBC Connector Sink を使用する # JDBC コネクタ このコネクタは、データがシンプルで、int などのプリミティブなデータ型だけで構成されている場合にのみ使用してください。Map などの ClickHouse 固有の型はサポートされていません。 この例では、Confluent の Kafka Connect ディストリビューションを使用します。 以下では、単一の Kafka トピックからメッセージを取得し、ClickHouse テーブルに行を挿入するシンプルなインストール方法を説明します。Kafka 環境がない場合は、無料 tier が充実している Confluent Cloud を推奨します。 JDBC コネクタではスキーマが必須である点に注意してください (JDBC コネクタではプレーンな JSON や CSV は使用できません) 。スキーマは各メッセージにエンコードすることもできますが、その場合のオーバーヘッドを避けるため、[Confluent スキーマレジストリの使用を強く推奨します](https://www.confluent.io/blog/kafka-connect-deep-dive-converters-serialization-explained/#json-schemas)。提供されている挿入スクリプトは、メッセージから自動的にスキーマを推論してレジストリに登録するため、他のデータセットにも再利用できます。Kafka のキーは String であることを前提としています。Kafka のスキーマの詳細は、[こちら](https://docs.confluent.io/platform/current/schema-registry/index.html)を参照してください。
### ライセンス
JDBC コネクタは、[Confluent Community License](https://www.confluent.io/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** をクリックします。
ClickHouse Cloud サービスの接続ボタン
**HTTPS** を選択します。接続情報は `curl` コマンドの例として表示されます。
ClickHouse Cloud HTTPS 接続情報
セルフマネージド ClickHouse を使用している場合、接続情報は ClickHouse 管理者によって設定されます。 Confluent パッケージをダウンロードし、ローカル環境にインストール済みであることを前提としています。コネクタのインストールについては、[こちら](https://docs.confluent.io/kafka-connect-jdbc/current/#install-the-jdbc-connector)に記載されている手順に従ってください。 `confluent-hub` を使用してインストールする場合、ローカルの設定ファイルが更新されます。 Kafka から ClickHouse にデータを送信するには、コネクタの Sink コンポーネントを使用します。 [こちら](https://github.com/ClickHouse/clickhouse-java/releases)から ClickHouse JDBCドライバー `clickhouse-jdbc--shaded.jar` をダウンロードしてインストールしてください。Kafka Connect へのインストールは、[こちら](https://docs.confluent.io/kafka-connect-jdbc/current/#installing-jdbc-drivers)の手順に従ってください。他のドライバーでも動作する可能性はありますが、テストは行っていません。 よくある問題: ドキュメントでは jar を `share/java/kafka-connect-jdbc/` にコピーするよう案内されています。Connect がドライバーを見つけられない場合は、ドライバーを `share/confluent-hub-components/confluentinc-kafka-connect-jdbc/lib/` にコピーしてください。あるいは、`plugin.path` を変更してドライバーを含めてください。詳細は以下を参照してください。 インストール形態に応じた Connect の設定については、スタンドアロン クラスターと 分散 クラスターの違いに注意しつつ、[こちらの手順](https://docs.confluent.io/cloud/current/cp-component/connect-cloud-config.html#set-up-a-local-connect-worker-with-cp-install)に従ってください。Confluent Cloud を使用する場合は、分散 構成が該当します。 以下のパラメータは、ClickHouse で JDBC コネクタを使用する際に重要です。パラメータの完全な一覧は[こちら](https://docs.confluent.io/kafka-connect-jdbc/current/sink-connector/index.html)で確認できます。 * `_connection.url_` - `jdbc:clickhouse://<clickhouse host>:<clickhouse http port>/<target database>` の形式で指定する必要があります * `connection.user` - 移行先データベースへの書き込み権限を持つユーザー * `table.name.format`- データの挿入先となる ClickHouse テーブル。事前に存在している必要があります。 * `batch.size` - 1 回のバッチで送信する行数です。適切に大きな値を設定してください。ClickHouse の[推奨事項](/docs/ja/reference/statements/insert-into#performance-considerations)では、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 のサンプルデータ用の設定ファイル例は[こちら](https://github.com/ClickHouse/kafka-samples/tree/main/github_events/jdbc_sink)にあります。これは、Connect を スタンドアロン モードで実行し、Kafka を Confluent Cloud でホストしていることを前提としています。 テーブルが作成されていることを確認し、以前の例ですでに存在している場合は削除してください。縮小版の GitHub dataset に対応した例を以下に示します。現在サポートされていない Array 型や Map 型が含まれていない点に注意してください: ```sql theme={null} 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 を [スタンドアロン](https://docs.confluent.io/cloud/current/cp-component/connect-cloud-config.html#standalone-cluster) モードまたは [分散](https://docs.confluent.io/cloud/current/cp-component/connect-cloud-config.html#distributed-cluster) モードで起動します。 ```bash theme={null} ./bin/connect-standalone connect.properties.ini github-jdbc-sink.properties.ini ``` 提供されている[スクリプトと設定](https://github.com/ClickHouse/kafka-samples/tree/main/producer)を使用して、Kafka にメッセージを送信します。`github.config` を変更して、Kafka の認証情報を含める必要があります。このスクリプトは現在、Confluent Cloud で使用するように設定されています。 ```bash theme={null} python producer.py -c github.config ``` このスクリプトを使うと、任意の ndjson ファイルを Kafka topic に挿入できます。スキーマの自動推論も試みます。提供されているサンプル config では 10k 件のメッセージしか挿入されません。必要に応じて[こちらで変更](https://github.com/ClickHouse/clickhouse-docs/tree/main/docs/integrations/data-ingestion/kafka/code/producer/github.config#L25)してください。この設定では、Kafka への挿入時にデータセットから互換性のない Array フィールドも削除されます。 これは、JDBC コネクタがメッセージを INSERT 文に変換するために必要です。独自のデータを使用する場合は、すべてのメッセージにスキーマを含めて挿入する (\_value.converter.schemas.enable \_を true に設定する) か、client がスキーマレジストリ内のスキーマを参照するメッセージを公開するようにしてください。 Kafka Connect はメッセージの消費を開始し、ClickHouse への行の挿入を開始するはずです。"\[JDBC Compliant Mode] Transaction isn't supported." に関する警告は想定内であり、無視してかまいません。 ターゲットテーブル "Github" を単純に読み取れば、データが挿入されたことを確認できるはずです。 ```sql theme={null} SELECT count() FROM default.github; ``` ```response theme={null} | count\(\) | | :--- | | 10000 | ``` * [Kafka Sink の設定パラメータ](https://docs.confluent.io/kafka-connect-jdbc/current/sink-connector/sink_config_options.html#sink-config-options) * [Kafka Connect 詳説 – JDBC ソースコネクタ](https://www.confluent.io/blog/kafka-connect-deep-dive-jdbc-source-connector) * [Kafka Connect JDBC Sink 詳説: 主キーの扱い](https://rmoff.net/2021/03/12/kafka-connect-jdbc-sink-deep-dive-working-with-primary-keys/) * [Kafka Connect 実践: JDBC Sink](https://www.youtube.com/watch?v=b-3qN_tlYR4\&t=981s) - 読むより視聴したい方はこちら。 * [Kafka Connect 詳説 – コンバータとシリアライゼーションの解説](https://www.confluent.io/blog/kafka-connect-deep-dive-converters-serialization-explained/#json-schemas)