> ## Documentation Index
> Fetch the complete documentation index at: https://private-7c7dfe99-mintlify-8c05c8a2.mintlify.site/llms.txt
> Use this file to discover all available pages before exploring further.

> Kafka Connect 및 ClickHouse에서 JDBC 커넥터 싱크 사용

# JDBC 커넥터

<Note>
  이 커넥터는 데이터가 단순하고 `int`와 같은 기본 데이터 타입으로만 구성된 경우에만 사용해야 합니다. 맵과 같은 ClickHouse 고유 타입은 지원되지 않습니다.
</Note>

예시에서는 Kafka Connect의 Confluent 배포판을 사용합니다.

아래에서는 단일 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)y를 사용하는 것을 강력히 권장합니다. 제공되는 삽입 스크립트는 메시지에서 스키마를 자동으로 추론해 레지스트리에 삽입하므로, 다른 데이터셋에도 재사용할 수 있습니다. Kafka의 keys는 String이라고 가정합니다. Kafka 스키마에 대한 자세한 내용은 [여기](https://docs.confluent.io/platform/current/schema-registry/index.html)에서 확인할 수 있습니다.

<div id="license">
  ### 라이선스
</div>

JDBC 커넥터는 [Confluent Community License](https://www.confluent.io/confluent-community-license)에 따라 배포됩니다.

<div id="steps">
  ### 절차
</div>

<div id="gather-your-connection-details">
  #### 연결 정보를 확인합니다
</div>

HTTP(S)로 ClickHouse에 연결하려면 다음 정보가 필요합니다.

| 매개변수                      | 설명                                                         |
| ------------------------- | ---------------------------------------------------------- |
| `HOST` and `PORT`         | 일반적으로 TLS를 사용하는 경우 포트는 8443, TLS를 사용하지 않는 경우 8123입니다.      |
| `DATABASE NAME`           | 기본적으로 `default`라는 이름의 데이터베이스가 제공되며, 연결할 데이터베이스 이름을 사용하십시오. |
| `USERNAME` and `PASSWORD` | 기본 사용자 이름은 `default`입니다. 사용 사례에 맞는 사용자 이름을 사용하십시오.         |

ClickHouse Cloud 서비스의 연결 정보는 ClickHouse Cloud 콘솔에서 확인할 수 있습니다.
서비스를 선택한 다음 **Connect**를 클릭하십시오.

<Image img="/images/_snippets/cloud-connect-button.png" size="md" alt="ClickHouse Cloud 서비스 연결 버튼" border />

**HTTPS**를 선택하십시오. 연결 정보가 예시 `curl` 명령으로 표시됩니다.

<Image img="/images/_snippets/connection-details-https.png" size="md" alt="ClickHouse Cloud HTTPS 연결 정보" border />

자가 관리형 ClickHouse를 사용하는 경우 연결 정보는 ClickHouse 관리자가 설정합니다.

<div id="1-install-kafka-connect-and-connector">
  #### 1. Kafka Connect 및 커넥터 설치
</div>

Confluent 패키지를 다운로드하여 로컬에 설치했다고 가정합니다. 커넥터를 설치하려면 [여기](https://docs.confluent.io/kafka-connect-jdbc/current/#install-the-jdbc-connector)에 안내된 설치 지침을 따르십시오.

`confluent-hub` 설치 방식을 사용하는 경우 로컬 설정 파일이 업데이트됩니다.

Kafka에서 ClickHouse로 데이터를 전송하기 위해 커넥터의 싱크 구성 요소를 사용합니다.

<div id="2-download-and-install-the-jdbc-driver">
  #### 2. JDBC 드라이버 다운로드 및 설치
</div>

[여기](https://github.com/ClickHouse/clickhouse-java/releases)에서 ClickHouse JDBC 드라이버 `clickhouse-jdbc-<version>-shaded.jar`를 다운로드하여 설치하십시오. [여기](https://docs.confluent.io/kafka-connect-jdbc/current/#installing-jdbc-drivers)의 안내에 따라 Kafka Connect에 설치하십시오. 다른 드라이버도 작동할 수 있지만 테스트되지는 않았습니다.

<Note>
  일반적인 문제: 문서에서는 jar 파일을 `share/java/kafka-connect-jdbc/`로 복사하라고 안내합니다. Connect가 드라이버를 찾지 못하면 드라이버를 `share/confluent-hub-components/confluentinc-kafka-connect-jdbc/lib/`로 복사하십시오. 또는 드라이버가 포함되도록 `plugin.path`를 수정하십시오. 자세한 내용은 아래를 참조하십시오.
</Note>

<div id="3-prepare-configuration">
  #### 3. 구성 준비
</div>

설치 유형에 맞는 Connect를 설정하려면 [이 지침](https://docs.confluent.io/cloud/current/cp-component/connect-cloud-config.html#set-up-a-local-connect-worker-with-cp-install)을 따르십시오. standalone cluster와 distributed cluster의 차이점에 유의해야 합니다. Confluent Cloud를 사용하는 경우에는 distributed 구성이 해당됩니다.

다음 매개변수는 ClickHouse와 함께 JDBC 커넥터를 사용할 때 중요합니다. 전체 매개변수 목록은 [여기](https://docs.confluent.io/kafka-connect-jdbc/current/sink-connector/index.html)에서 확인할 수 있습니다.

* `_connection.url_` - `jdbc:clickhouse://&lt;clickhouse host>:&lt;clickhouse http port>/&lt;target database>` 형식이어야 합니다.
* `connection.user` - 대상 데이터베이스에 쓰기 권한이 있는 사용자
* `table.name.format`- 데이터를 삽입할 ClickHouse 테이블입니다. 이 테이블은 반드시 미리 존재해야 합니다.
* `batch.size` - 한 번의 batch로 전송할 행 수입니다. 이 값은 충분히 크게 설정해야 합니다. ClickHouse [권장 사항](/ko/reference/statements/insert-into#performance-considerations)에 따르면 1000은 최소값으로 보는 것이 좋습니다.
* `tasks.max` - JDBC 싱크 커넥터는 하나 이상의 작업을 실행할 수 있습니다. 이를 통해 성능을 높일 수 있습니다. batch 크기와 함께 성능을 개선하는 주요 수단입니다.
* `value.converter.schemas.enable` - 스키마 레지스트리를 사용하는 경우 false로, 스키마를 메시지에 포함하는 경우 true로 설정합니다.
* `value.converter` - 데이터 타입에 맞게 설정합니다. 예를 들어 JSON의 경우 `io.confluent.connect.json.JsonSchemaConverter`를 사용합니다.
* `key.converter` - `org.apache.kafka.connect.storage.StringConverter`로 설정합니다. String 키를 사용합니다.
* `pk.mode` - ClickHouse에서는 관련이 없습니다. none으로 설정합니다.
* `auto.create` - 지원되지 않으므로 반드시 false로 설정해야 합니다.
* `auto.evolve` - 현재는 false로 설정하는 것을 권장하지만, 향후 지원될 수 있습니다.
* `insert.mode` - "insert"로 설정합니다. 현재 다른 모드는 지원되지 않습니다.
* `key.converter` - 키의 타입에 맞게 설정합니다.
* `value.converter` - topic의 데이터 타입에 따라 설정합니다. 이 데이터는 지원되는 스키마를 가져야 하며, JSON, Avro 또는 Protobuf 포맷이어야 합니다.

테스트에 샘플 데이터셋을 사용하는 경우 다음 설정을 확인하십시오.

* `value.converter.schemas.enable` - 스키마 레지스트리를 사용하므로 false로 설정합니다. 각 메시지에 스키마를 포함하는 경우 true로 설정합니다.
* `key.converter` - "org.apache.kafka.connect.storage.StringConverter"로 설정합니다. String 키를 사용합니다.
* `value.converter` - "io.confluent.connect.json.JsonSchemaConverter"로 설정합니다.
* `value.converter.schema.registry.url` - schema server URL로 설정하고, `value.converter.schema.registry.basic.auth.user.info` 매개변수로 schema server의 자격 증명도 함께 설정합니다.

Github 샘플 데이터용 예시 설정 파일은 [여기](https://github.com/ClickHouse/kafka-samples/tree/main/github_events/jdbc_sink)에서 확인할 수 있으며, Connect가 standalone 모드로 실행되고 Kafka가 Confluent Cloud에서 호스팅된다고 가정합니다.

<div id="4-create-the-clickhouse-table">
  #### 4. ClickHouse 테이블 생성
</div>

테이블이 생성되었는지 확인하고, 이전 예시로 인해 이미 존재하는 경우에는 삭제하십시오. 축소된 Github 데이터셋과 호환되는 예시는 아래와 같습니다. 현재 지원되지 않는 배열 또는 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)
```

<div id="5-start-kafka-connect">
  #### 5. Kafka Connect 시작
</div>

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
```

<div id="6-add-data-to-kafka">
  #### 6. Kafka에 데이터 추가
</div>

제공된 [스크립트와 설정](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에 삽입하는 데 사용할 수 있습니다. 이 과정에서 schema를 자동으로 추론합니다. 제공된 예제 구성은 10k개의 메시지만 삽입합니다. 필요하면 [여기에서 수정](https://github.com/ClickHouse/clickhouse-docs/tree/main/docs/integrations/data-ingestion/kafka/code/producer/github.config#L25)하십시오. 또한 이 구성은 Kafka에 삽입하는 동안 데이터셋에서 호환되지 않는 배열(Array) field를 모두 제거합니다.

이는 JDBC 커넥터가 메시지를 INSERT SQL 문으로 변환하는 데 필요합니다. 자체 데이터를 사용하는 경우, 모든 메시지에 schema를 함께 삽입하거나 (`_value.converter.schemas.enable`을 true로 설정) 클라이언트가 registry의 schema를 참조하는 메시지를 게시하도록 해야 합니다.

Kafka Connect가 메시지 소비를 시작하고 ClickHouse에 행을 삽입하기 시작합니다. "\[JDBC Compliant Mode] Transaction isn't supported." 관련 경고는 예상되는 것이므로 무시해도 됩니다.

대상 테이블 "Github"를 간단히 조회하여 데이터 삽입 여부를 확인할 수 있습니다.

```sql theme={null}
SELECT count() FROM default.github;
```

```response theme={null}
| count\(\) |
| :--- |
| 10000 |
```

<div id="recommended-further-reading">
  ### 추가로 읽어볼 자료
</div>

* [Kafka 싱크 구성 매개변수](https://docs.confluent.io/kafka-connect-jdbc/current/sink-connector/sink_config_options.html#sink-config-options)
* [Kafka Connect 심층 분석 – JDBC Source Connector](https://www.confluent.io/blog/kafka-connect-deep-dive-jdbc-source-connector)
* [Kafka Connect JDBC 싱크 심층 분석: 기본 키(Primary Keys) 다루기](https://rmoff.net/2021/03/12/kafka-connect-jdbc-sink-deep-dive-working-with-primary-keys/)
* [Kafka Connect in Action: JDBC 싱크](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)