` 内のセクションには、Kafka の拡張設定オプションが含まれています。その他のオプションについては、[librdkafka の設定](https://github.com/confluentinc/librdkafka/blob/master/CONFIGURATION.md)を参照してください。
4. この例では、`SASL_SSL` セキュリティプロトコルと `PLAIN` メカニズムを使用しています。これらの設定は、Kafka クラスターの構成に応じて調整してください。
## テーブルとデータベースの作成
ClickHouseクラスター上に必要なデータベースとテーブルを作成します。ClickHouseを単一ノードで実行している場合は、SQLコマンドのクラスター指定部分を省略し、`ReplicatedMergeTree` の代わりに別のエンジンを使用してください。
### データベースを作成する
```sql theme={null}
CREATE DATABASE kafka_testing ON CLUSTER LAB_CLICKHOUSE_CLUSTER;
```
### Kafkaテーブルを作成する
1つ目のKafkaクラスター用に、1つ目のKafkaテーブルを作成します。
```sql theme={null}
CREATE TABLE kafka_testing.first_kafka_table ON CLUSTER LAB_CLICKHOUSE_CLUSTER
(
`id` UInt32,
`first_name` String,
`last_name` String
)
ENGINE = Kafka(cluster_1);
```
2 つ目の Kafka クラスター用に、2 つ目の Kafka テーブルを作成します。
```sql theme={null}
CREATE TABLE kafka_testing.second_kafka_table ON CLUSTER STAGE_CLICKHOUSE_CLUSTER
(
`id` UInt32,
`first_name` String,
`last_name` String
)
ENGINE = Kafka(cluster_2);
```
### レプリケートテーブルを作成する
最初のKafkaテーブル用に、次のテーブルを作成します：
```sql theme={null}
CREATE TABLE kafka_testing.first_replicated_table ON CLUSTER STAGE_CLICKHOUSE_CLUSTER
(
`id` UInt32,
`first_name` String,
`last_name` String
) ENGINE = ReplicatedMergeTree()
ORDER BY id;
```
2つ目のKafkaテーブル用のテーブルを作成します:
```sql theme={null}
CREATE TABLE kafka_testing.second_replicated_table ON CLUSTER STAGE_CLICKHOUSE_CLUSTER
(
`id` UInt32,
`first_name` String,
`last_name` String
) ENGINE = ReplicatedMergeTree()
ORDER BY id;
```
### materialized view を作成する
最初の Kafka テーブルから最初のレプリケートテーブルへデータを挿入する materialized view を作成します。
```sql theme={null}
CREATE MATERIALIZED VIEW kafka_testing.cluster_1_mv ON CLUSTER STAGE_CLICKHOUSE_CLUSTER TO first_replicated_table AS
SELECT
id,
first_name,
last_name
FROM first_kafka_table;
```
2 つ目の Kafka テーブルから 2 つ目のレプリケートテーブルへデータを挿入するための materialized view を作成します。
```sql theme={null}
CREATE MATERIALIZED VIEW kafka_testing.cluster_2_mv ON CLUSTER STAGE_CLICKHOUSE_CLUSTER TO second_replicated_table AS
SELECT
id,
first_name,
last_name
FROM second_kafka_table;
```
## セットアップの確認
これで、Kafkaクラスター上に対応するコンシューマグループが表示されているはずです。
* `cluster_1_clickhouse_consumer` (`cluster_1` 上)
* `cluster_2_clickhouse_consumer` (`cluster_2` 上)
両方のテーブルのデータを確認するには、いずれかの ClickHouse ノードで次のクエリを実行します。
```sql theme={null}
SELECT * FROM first_replicated_table LIMIT 10;
```
```sql theme={null}
SELECT * FROM second_replicated_table LIMIT 10;
```
### 注
このガイドでは、両方のKafkaトピックに取り込まれるデータは同一です。実際の環境では、異なるデータになるはずです。Kafkaクラスターは必要な数だけ追加できます。
出力例:
```response theme={null}
┌─id─┬─first_name─┬─last_name─┐
│ 0 │ FirstName0 │ LastName0 │
│ 1 │ FirstName1 │ LastName1 │
│ 2 │ FirstName2 │ LastName2 │
└────┴────────────┴───────────┘
```
これで、named collections を使用して ClickHouse と Kafka を統合するためのセットアップは完了です。Kafka の設定を ClickHouse の `config.xml` ファイルに集約することで、設定の管理や調整が容易になり、よりシンプルで効率的なインテグレーションを実現できます。