ClickHouse에 dlt 연결

dlt는 다양한, 그리고 대개 정리가 덜 된 데이터 소스의 데이터를 구조가 잘 잡힌 실시간 데이터셋으로 적재할 수 있도록 Python 스크립트에 추가하는 오픈소스 라이브러리입니다.

pip install "dlt[clickhouse]"

dlt init chess clickhouse

pip install -r requirements.txt

CREATE DATABASE IF NOT EXISTS dlt;
CREATE USER dlt IDENTIFIED WITH sha256_password BY 'Dlt*12345789234567';
GRANT CREATE, ALTER, SELECT, DELETE, DROP, TRUNCATE, OPTIMIZE, SHOW, INSERT, dictGet ON dlt.* TO dlt;
GRANT SELECT ON INFORMATION_SCHEMA.COLUMNS TO dlt;
GRANT CREATE TEMPORARY TABLE, S3 ON *.* TO dlt;

[destination.clickhouse.credentials]
database = "dlt"                         # 생성한 데이터베이스 이름
username = "dlt"                         # ClickHouse 사용자 이름, 기본값은 일반적으로 "default"
password = "Dlt*12345789234567"          # ClickHouse 비밀번호(있는 경우)
host = "localhost"                       # ClickHouse 서버 호스트
port = 9000                              # ClickHouse 포트, 기본값은 9000
http_port = 8443                         # ClickHouse 서버의 HTTP 인터페이스에 연결할 HTTP 포트입니다. 기본값은 8443입니다.
secure = 1                               # HTTPS를 사용하는 경우 1, 그렇지 않으면 0으로 설정합니다.

[destination.clickhouse]
dataset_table_separator = "___"          # 데이터셋에서 생성되는 테이블 이름에 사용할 구분자입니다.

# toml 파일의 맨 위에, 어떤 섹션보다 앞에 두십시오.
destination.clickhouse.credentials="clickhouse://dlt:Dlt*12345789234567@localhost:9000/dlt?secure=1"

모든 write dispositions 을 지원합니다.

dlt 라이브러리에서 write disposition은 데이터를 대상에 어떤 방식으로 기록할지 정의합니다. write disposition에는 세 가지 유형이 있습니다.

Replace: 이 방식은 대상의 데이터를 리소스의 데이터로 대체합니다. 모든 클래스와 객체를 삭제하고, 데이터를 로드하기 전에 스키마를 다시 생성합니다. 자세한 내용은 여기에서 확인할 수 있습니다.

Merge: 이 쓰기 방식은 리소스의 데이터를 대상의 기존 데이터와 병합합니다. merge 방식을 사용하려면 리소스에 primary_key를 지정해야 합니다. 자세한 내용은 여기에서 확인할 수 있습니다.

Append: 기본 방식입니다. primary_key 필드는 무시하고 데이터를 대상의 기존 데이터에 추가합니다.

데이터는 데이터 소스에 따라 가장 효율적인 방식으로 ClickHouse에 로드됩니다:

• 로컬 파일의 경우 clickhouse-connect 라이브러리를 사용해 INSERT 명령으로 파일을 ClickHouse 테이블에 직접 로드합니다.
• S3, Google Cloud Storage, Azure Blob Storage와 같은 원격 스토리지의 파일은 s3, gcs, azureBlobStorage와 같은 ClickHouse 테이블 함수를 사용해 읽은 뒤, 데이터를 테이블에 삽입합니다.

ClickHouse는 하나의 데이터베이스에서 여러 데이터셋을 지원하지 않지만, dlt는 여러 이유로 데이터셋에 의존합니다. ClickHouse를 dlt와 함께 사용할 수 있도록, ClickHouse 데이터베이스에서 dlt가 생성한 테이블 이름 앞에는 구성 가능한 dataset_table_separator로 구분된 데이터셋 이름 접두사가 붙습니다. 또한 아무 데이터도 포함하지 않는 특수한 sentinel 테이블이 생성되며, 이를 통해 dlt는 ClickHouse 대상에 어떤 가상 데이터셋이 이미 존재하는지 식별할 수 있습니다.

• jsonl은 직접 로딩과 스테이징 모두에 권장되는 포맷입니다.
• parquet은 직접 로딩과 스테이징 모두에서 지원됩니다.

clickhouse 대상에는 기본 SQL 대상과 다른 몇 가지 고유한 차이점이 있습니다:

1. ClickHouse에는 실험적인 object 데이터 유형(datatype)이 있지만, 다소 예측 불가능하게 동작하는 것으로 확인되었습니다. 따라서 dlt clickhouse 대상은 복합 데이터 유형을 텍스트 컬럼으로 로드합니다. 이 기능이 필요하다면 Slack 커뮤니티에 문의해 주십시오. 추가 지원을 검토하겠습니다.
2. ClickHouse는 time 데이터 유형을 지원하지 않습니다. time은 text 컬럼으로 로드됩니다.
3. ClickHouse는 binary 데이터 유형을 지원하지 않습니다. 대신 바이너리 데이터는 text 컬럼으로 로드됩니다. jsonl에서 로드할 때 바이너리 데이터는 base64 문자열이 되며, parquet에서 로드할 때는 binary 객체가 text로 변환됩니다.
4. ClickHouse는 이미 데이터가 들어 있는 테이블에 NULL이 아닌 컬럼을 추가할 수 있습니다.
5. ClickHouse는 float 또는 double 데이터 유형을 사용할 때 특정 조건에서 반올림 오류를 일으킬 수 있습니다. 반올림 오류가 허용되지 않는 경우에는 반드시 decimal 데이터 유형을 사용하십시오. 예를 들어, 로더 파일 포맷을 jsonl로 설정한 상태에서 값 12.7001을 double 컬럼에 로드하면 예측 가능한 반올림 오류가 발생합니다.

ClickHouse는 다음 컬럼 힌트를 지원합니다.

• primary_key - 해당 컬럼을 프라이머리 키의 일부로 지정합니다. 여러 컬럼에 이 힌트를 지정해 복합 프라이머리 키를 만들 수 있습니다.

기본적으로 테이블은 ClickHouse에서 ReplicatedMergeTree 테이블 엔진으로 생성됩니다. clickhouse 어댑터에서 table_engine_type을 사용해 다른 테이블 엔진을 지정할 수 있습니다:

from dlt.destinations.adapters import clickhouse_adapter

@dlt.resource()
def my_resource():
  ...

clickhouse_adapter(my_resource, table_engine_type="merge_tree")

지원되는 값은 다음과 같습니다.

• merge_tree - MergeTree 엔진을 사용해 테이블을 생성합니다
• replicated_merge_tree (기본값) - ReplicatedMergeTree 엔진을 사용해 테이블을 생성합니다

ClickHouse는 Amazon S3, Google Cloud Storage, Azure Blob Storage를 파일 스테이징 대상으로 지원합니다.

dlt는 Parquet 또는 jsonl 파일을 스테이징 위치에 업로드한 다음, ClickHouse 테이블 함수를 사용해 스테이징된 파일에서 직접 데이터를 로드합니다.

스테이징 대상에 사용할 자격 증명을 구성하는 방법은 파일 시스템(filesystem) 문서를 참조하십시오.

• Amazon S3
• Google Cloud Storage
• Azure Blob Storage

스테이징을 활성화한 상태로 파이프라인을 실행하려면:

pipeline = dlt.pipeline(
  pipeline_name='chess_pipeline',
  destination='clickhouse',
  staging='filesystem',  # 스테이징을 활성화하려면 추가하세요
  dataset_name='chess_data'
)

dlt는 데이터를 ClickHouse에 적재할 때 스테이징 영역으로 Google Cloud Storage(GCS)를 사용할 수 있습니다. 이는 dlt가 내부적으로 사용하는 ClickHouse의 GCS 테이블 함수를 통해 자동으로 처리됩니다.

ClickHouse GCS 테이블 함수는 Hash-based Message Authentication Code(HMAC) 키를 사용한 인증만 지원합니다. 이를 위해 GCS는 Amazon S3 API를 에뮬레이션하는 S3 호환 모드를 제공합니다. ClickHouse는 이를 활용해 S3 통합을 통해 GCS 버킷에 접근할 수 있습니다.

dlt에서 HMAC 인증으로 GCS 스테이징을 설정하려면 다음과 같이 하십시오:

1. Google Cloud 가이드에 따라 GCS 서비스 계정의 HMAC 키를 생성하십시오.

2. dlt 프로젝트의 config.toml에 있는 ClickHouse 대상 설정에서 서비스 계정의 HMAC 키와 client_email, project_id, private_key를 구성하십시오:

[destination.filesystem]
bucket_url = "gs://dlt-ci"

[destination.filesystem.credentials]
project_id = "a-cool-project"
client_email = "my-service-account@a-cool-project.iam.gserviceaccount.com"
private_key = "-----BEGIN PRIVATE KEY-----\nMIIEvQIBADANBgkaslkdjflasjnkdcopauihj...wEiEx7y+mx\nNffxQBqVVej2n/D93xY99pM=\n-----END PRIVATE KEY-----\n"

[destination.clickhouse.credentials]
database = "dlt"
username = "dlt"
password = "Dlt*12345789234567"
host = "localhost"
port = 9440
secure = 1
gcp_access_key_id = "JFJ$$*f2058024835jFffsadf"
gcp_secret_access_key = "DFJdwslf2hf57)%$02jaflsedjfasoi"

참고: HMAC 키 bashgcp_access_key_id 및 gcp_secret_access_key)외에도 이제[destination.filesystem.credentials]아래에 서비스 계정의client_email, project_id, private_key`를 제공해야 합니다. 이는 GCS 스테이징 지원이 현재 임시 우회책으로 구현되어 있고 아직 최적화되지 않았기 때문입니다.

dlt는 이러한 자격 증명을 ClickHouse에 전달하며, ClickHouse가 인증 및 GCS 액세스를 처리합니다.

향후 ClickHouse dlt 대상의 GCS 스테이징 설정을 더 단순하고 개선된 방식으로 제공하기 위한 작업이 활발히 진행 중입니다. 정식 GCS 스테이징 지원은 다음 GitHub 이슈에서 추적하고 있습니다.

• 파일 시스템 대상이 gcs의 s3 호환 모드에서 작동하도록 지원
• Google Cloud Storage 스테이징 영역 지원

dbt 통합은 일반적으로 dbt-clickhouse를 통해 지원됩니다.

이 대상은 dlt 상태 동기화를 완벽하게 지원합니다.