JDBC 커넥터 - ClickHouse Documentation

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

예시에서는 Kafka Connect의 Confluent 배포판을 사용합니다. 아래에서는 단일 Kafka 토픽에서 메시지를 가져와 ClickHouse 테이블에 행을 삽입하는 간단한 설치 방법을 설명합니다. Kafka 환경이 없는 경우에는 넉넉한 무료 tier를 제공하는 Confluent Cloud를 권장합니다. JDBC 커넥터를 사용하려면 스키마가 필요합니다(JDBC 커넥터에서는 일반 JSON 또는 CSV를 사용할 수 없습니다). 스키마를 각 메시지에 인코딩할 수도 있지만, 그에 따른 오버헤드를 피하기 위해 Confluent 스키마 레지스트리y를 사용하는 것을 강력히 권장합니다. 제공되는 삽입 스크립트는 메시지에서 스키마를 자동으로 추론해 레지스트리에 삽입하므로, 다른 데이터셋에도 재사용할 수 있습니다. Kafka의 keys는 String이라고 가정합니다. Kafka 스키마에 대한 자세한 내용은 여기에서 확인할 수 있습니다.

라이선스

JDBC 커넥터는 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를 클릭하십시오.

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

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

Kafka Connect 및 커넥터 설치

Confluent 패키지를 다운로드하여 로컬에 설치했다고 가정합니다. 커넥터를 설치하려면 여기에 설명된 설치 지침을 따르십시오.confluent-hub 설치 방법을 사용하는 경우 로컬 설정 파일이 업데이트됩니다.Kafka에서 ClickHouse로 데이터를 전송하려면 커넥터의 싱크 컴포넌트를 사용합니다.

JDBC 드라이버 다운로드 및 설치

ClickHouse JDBC 드라이버 clickhouse-jdbc-<version>-shaded.jar를 여기에서 다운로드하여 설치하십시오. 여기에 안내된 내용에 따라 Kafka Connect에 설치하십시오. 다른 드라이버도 작동할 수 있지만 테스트되지는 않았습니다.

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

구성 준비

설치 유형에 맞는 Connect를 설정하려면 이 지침을 따르십시오. standalone cluster와 distributed cluster의 차이점에 유의해야 합니다. Confluent Cloud를 사용하는 경우에는 distributed 구성이 해당됩니다.다음 매개변수는 ClickHouse와 함께 JDBC 커넥터를 사용할 때 중요합니다. 전체 매개변수 목록은 여기에서 확인할 수 있습니다.

_connection.url_ - jdbc:clickhouse://<clickhouse host>:<clickhouse http port>/<target database> 형식이어야 합니다.
connection.user - 대상 데이터베이스에 쓰기 권한이 있는 사용자
table.name.format- 데이터를 삽입할 ClickHouse 테이블입니다. 이 테이블은 반드시 미리 존재해야 합니다.
batch.size - 한 번의 batch로 전송할 행 수입니다. 이 값은 충분히 크게 설정해야 합니다. ClickHouse 권장 사항에 따르면 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 샘플 데이터용 예시 설정 파일은 여기에서 확인할 수 있으며, Connect가 standalone 모드로 실행되고 Kafka가 Confluent Cloud에서 호스팅된다고 가정합니다.

ClickHouse 테이블 생성

테이블이 생성되었는지 확인하고, 이전 예시에서 이미 존재하는 경우 삭제하십시오. 축소된 Github 데이터셋과 호환되는 예시는 아래와 같습니다. 현재 지원되지 않는 Array 또는 Map 타입이 없다는 점에 유의하십시오:

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 시작하기

Kafka Connect를 standalone 또는 distributed 모드로 시작하십시오.

./bin/connect-standalone connect.properties.ini github-jdbc-sink.properties.ini

Kafka에 데이터 추가하기

제공된 스크립트와 구성을 사용하여 Kafka에 메시지를 전송하세요. github.config에 Kafka 자격 증명을 포함하도록 수정해야 합니다. 이 스크립트는 현재 Confluent Cloud에서 사용하도록 구성되어 있습니다.

python producer.py -c github.config

이 스크립트는 임의의 ndjson 파일을 Kafka topic에 삽입하는 데 사용할 수 있습니다. 스키마를 자동으로 추론합니다. 제공된 샘플 구성은 10k개의 메시지만 삽입합니다. 필요하면 여기에서 수정하십시오. 또한 이 구성은 Kafka에 삽입하는 동안 데이터셋에서 호환되지 않는 배열 필드를 제거합니다.이 작업은 JDBC 커넥터가 메시지를 INSERT SQL 문으로 변환하는 데 필요합니다. 자체 데이터를 사용하는 경우, 각 메시지와 함께 스키마를 삽입하거나(value.converter.schemas.enable을 true로 설정) 클라이언트가 레지스트리의 스키마를 참조하는 메시지를 게시하도록 해야 합니다.Kafka Connect가 메시지 소비를 시작하고 ClickHouse에 행을 삽입하기 시작해야 합니다. “[JDBC Compliant Mode] Transaction isn’t supported.”와 관련된 경고는 예상되는 동작이므로 무시해도 됩니다.대상 테이블 “Github”를 간단히 조회하면 데이터 삽입을 확인할 수 있습니다.

SELECT count() FROM default.github;

| count\(\) |
| :--- |
| 10000 |

추가로 읽어볼 자료

Kafka 싱크 구성 매개변수
Kafka Connect 심층 분석 – JDBC Source Connector
Kafka Connect JDBC 싱크 심층 분석: 기본 키(Primary Keys) 다루기
Kafka Connect in Action: JDBC 싱크 - 읽기보다 영상 시청을 선호하는 경우
Kafka Connect 심층 분석 – 컨버터와 직렬화 설명

​라이선스

​절차

​연결 정보를 확인합니다

​추가로 읽어볼 자료

라이선스

절차

연결 정보를 확인합니다

추가로 읽어볼 자료