` 내부의 섹션에는 확장 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 테이블 생성
첫 번째 Kafka 클러스터용 첫 번째 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);
```
두 번째 Kafka 클러스터용 두 번째 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);
```
### 복제된 테이블(Replicated Table) 생성
첫 번째 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;
```
두 번째 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 테이블의 데이터를 첫 번째 복제된 테이블(Replicated Table)에 삽입할 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;
```
두 번째 Kafka 테이블의 데이터를 두 번째 복제된 테이블에 삽입하도록 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 클러스터에서 해당 consumer group을 확인할 수 있습니다:
* `cluster_1`의 `cluster_1_clickhouse_consumer`
* `cluster_2`의 `cluster_2_clickhouse_consumer`
두 테이블의 데이터를 확인하려면 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 │
└────┴────────────┴───────────┘
```
이로써 이름이 지정된 컬렉션을 사용한 ClickHouse와 Kafka 통합 설정이 완료됩니다. Kafka 구성을 ClickHouse `config.xml` 파일에 중앙에서 관리하면 설정을 더 쉽게 관리하고 조정할 수 있어, 통합을 더욱 간편하고 효율적으로 운영할 수 있습니다.