> ## 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 커넥터 싱크 사용 # JDBC 커넥터 이 커넥터는 데이터가 단순하고 `int`와 같은 기본 데이터 타입으로만 구성된 경우에만 사용해야 합니다. 맵과 같은 ClickHouse 고유 타입은 지원되지 않습니다. 예시에서는 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)에서 확인할 수 있습니다.
### 라이선스
JDBC 커넥터는 [Confluent Community License](https://www.confluent.io/confluent-community-license)에 따라 배포됩니다.
### 절차
#### 연결 정보를 확인합니다
HTTP(S)로 ClickHouse에 연결하려면 다음 정보가 필요합니다. | 매개변수 | 설명 | | ------------------------- | ---------------------------------------------------------- | | `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로 데이터를 전송하려면 커넥터의 싱크 컴포넌트를 사용합니다. ClickHouse JDBC 드라이버 `clickhouse-jdbc--shaded.jar`를 [여기](https://github.com/ClickHouse/clickhouse-java/releases)에서 다운로드하여 설치하십시오. [여기](https://docs.confluent.io/kafka-connect-jdbc/current/#installing-jdbc-drivers)에 안내된 내용에 따라 Kafka Connect에 설치하십시오. 다른 드라이버도 작동할 수 있지만 테스트되지는 않았습니다. 일반적인 문제: 문서에서는 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)을 따르십시오. 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://<clickhouse host>:<clickhouse http port>/<target database>` 형식이어야 합니다. * `connection.user` - 대상 데이터베이스에 쓰기 권한이 있는 사용자 * `table.name.format`- 데이터를 삽입할 ClickHouse 테이블입니다. 이 테이블은 반드시 미리 존재해야 합니다. * `batch.size` - 한 번의 batch로 전송할 행 수입니다. 이 값은 충분히 크게 설정해야 합니다. ClickHouse [권장 사항](/docs/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에서 호스팅된다고 가정합니다. 테이블이 생성되었는지 확인하고, 이전 예시에서 이미 존재하는 경우 삭제하십시오. 축소된 Github 데이터셋과 호환되는 예시는 아래와 같습니다. 현재 지원되지 않는 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를 [standalone](https://docs.confluent.io/cloud/current/cp-component/connect-cloud-config.html#standalone-cluster) 또는 [distributed](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에 삽입하는 데 사용할 수 있습니다. 스키마를 자동으로 추론합니다. 제공된 샘플 구성은 10k개의 메시지만 삽입합니다. 필요하면 [여기에서 수정하십시오](https://github.com/ClickHouse/clickhouse-docs/tree/main/docs/integrations/data-ingestion/kafka/code/producer/github.config#L25). 또한 이 구성은 Kafka에 삽입하는 동안 데이터셋에서 호환되지 않는 배열 필드를 제거합니다. 이 작업은 JDBC 커넥터가 메시지를 INSERT SQL 문으로 변환하는 데 필요합니다. 자체 데이터를 사용하는 경우, 각 메시지와 함께 스키마를 삽입하거나(`value.converter.schemas.enable`을 true로 설정) 클라이언트가 레지스트리의 스키마를 참조하는 메시지를 게시하도록 해야 합니다. 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 싱크 구성 매개변수](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)