Delta 확장

Delta 확장 (Delta Extension)

delta 확장은 Delta Lake 오픈소스 스토리지 포맷에 대한 지원을 추가해요. Delta Kernel로 만들어졌어요. 이 확장은 Delta 테이블에 대한 읽기·쓰기 지원을 제공하며, 로컬과 원격 모두 지원해요.

구현 세부 사항은 [발표 블로그 글]({% post_url 2024-06-10-delta %})을 참고하세요.

Warning Azure Onelake에 delta-kernel-rs의 변경의 결과로 보이는 회귀(regression)가 있음을 알고 있어요. GitHub에서 이슈를 추적할 수 있어요.

Unity Catalog에 연결하려면 DuckDB에 [unity_catalog 확장]({% link docs/current/core_extensions/unity_catalog.md %})이 있어요.

출처: 문서

본문

설치와 로드

delta 확장은 첫 사용 시 공식 확장 저장소에서 투명하게 [자동 로드]({% link docs/current/extensions/overview.md %}#autoloading-extensions)돼요. 수동으로 설치·로드하고 싶다면:

INSTALL delta;
LOAD delta;

사용법

로컬 Delta 테이블을 스캔하려면:

SELECT *
FROM delta_scan('file:///some/path/on/local/machine');

S3 버킷에서 읽기

[S3 버킷]({% link docs/current/core_extensions/httpfs/s3api.md %})의 Delta 테이블을 스캔하려면:

SELECT *
FROM delta_scan('s3://some/delta/table');

S3 버킷 인증에는 DuckDB [Secrets]({% link docs/current/configuration/secrets_manager.md %})가 지원돼요.

CREATE SECRET (
    TYPE s3,
    PROVIDER credential_chain
);
SELECT *
FROM delta_scan('s3://some/delta/table/with/auth');

S3의 공개 버킷을 스캔하려면 공개 S3 버킷의 리전을 담은 secret을 만들어 올바른 리전을 넘겨야 할 수 있어요.

CREATE SECRET (
    TYPE s3,
    REGION 'my-region'
);
SELECT *
FROM delta_scan('s3://some/public/table/in/my-region');

Azure Blob Storage에서 읽기

[Azure Blob Storage 버킷]({% link docs/current/core_extensions/azure.md %}#azure-blob-storage)의 Delta 테이블을 스캔하려면:

SELECT *
FROM delta_scan('az://my-container/my-table');

Azure Blob Storage 인증에는 DuckDB [Secrets]({% link docs/current/configuration/secrets_manager.md %})가 지원돼요.

CREATE SECRET (
    TYPE azure,
    PROVIDER credential_chain
);
SELECT *
FROM delta_scan('az://my-container/my-table-with-auth');

Google Cloud Storage에서 읽기

[GCS 버킷]({% link docs/current/core_extensions/httpfs/s3api.md %})의 Delta 테이블을 스캔하려면 HMAC 키를 사용하고 secret을 만들어요.

CREATE SECRET (
    TYPE gcs,
    KEY_ID '⟨hmac-key-id⟩',
    SECRET '⟨hmac-secret⟩'
);
SELECT *
FROM delta_scan('gs://my-bucket/my-delta-table');

데이터 추가

Delta 테이블에 행을 추가하려면 부착하고 INSERT INTO를 사용해요.

ATTACH 's3://my-bucket/my-delta-table' AS my_table (TYPE delta);
INSERT INTO my_table SELECT * FROM other_table;

시간 여행 (Time Travel)

Delta 테이블의 특정 버전을 읽으려면 부착하고 AT (VERSION => n) 절을 사용해요.

ATTACH 's3://my-bucket/my-delta-table' AS my_table (TYPE delta);
SELECT * FROM my_table AT (VERSION => 5);

또는 부착 시점에 버전을 고정해요.

ATTACH 's3://my-bucket/my-delta-table' AS my_table (TYPE delta, VERSION 5);

체크포인트

부착된 테이블의 Delta 로그를 체크포인트 파일로 압축(compact)하려면:

ATTACH 'path/to/my-delta-table' AS my_table (TYPE delta);
CHECKPOINT my_table;

멱등 추가 (Idempotent Appends)

delta 확장은 Delta의 per-application 트랜잭션 버전을 사용해 exactly-once 의미론을 주는 멱등 추가 API를 노출해요. app_id와 버전으로 태그된 추가는 테이블이 현재 그 app_id에 대해 기록한 버전이 기대한 이전 버전과 일치할 때만 커밋돼요. 이렇게 하면 프로듀서가 배치를 중복 없이 안전하게 재시도할 수 있어요.

트랜잭션 안에서 delta_set_transaction_version(⟨table⟩, ⟨app_id⟩, ⟨new_version⟩, ⟨expected_previous_version⟩)로 추가를 태그해요.

ATTACH 'path/to/my-delta-table' AS my_table (TYPE delta);

BEGIN TRANSACTION;
CALL delta_set_transaction_version('my_table', 'my_app_id', 1::UBIGINT, NULL::UBIGINT);
INSERT INTO my_table VALUES (1);
COMMIT;

COMMIT 시 버전은 compare-and-swap돼요. 그 사이에 다른 프로세스가 my_app_id의 버전을 진행시켰다면 커밋이 실패해요. 트랜잭션을 중단하면 버전은 변경되지 않아요. 현재 버전은 delta_get_transaction_version(⟨table⟩, ⟨app_id⟩)로 읽는데, 아직 기록된 버전이 없으면 NULL을 반환해요.

부착 옵션

Delta 테이블을 부착할 때 ATTACH에 다음 옵션을 넘길 수 있어요.

ATTACH 's3://my-bucket/my-delta-table' AS my_table (
    TYPE delta,
    PIN_SNAPSHOT true,
    PUSHDOWN_FILTERS 'constant_only'
);

스캔된 파일 검사

delta_list_files는 스캔이 테이블에 대해 읽을 데이터 파일들을 카디널리티, 파티션 값, 삭제 벡터 보유 여부와 함께 반환해요. 이는 데이터 스킵핑의 효과를 이해하는 데 유용해요.

SELECT * FROM delta_list_files('file:///some/path/on/local/machine');

Delta의 자격 증명 체인

DuckDB Delta는 일부 네트워크 작업에 delta-kernel-rs와 object_store를 사용해요. 이 시스템들은 자격 증명 체인에 대해 다른 순서(및 포함 기본값)를 가져요. 시스템에 여러 자격 증명 소스가 있으면(예: 환경을 통한 Service Principal과 CLI 기반 옵션 둘 다), 자격 증명 로딩 동작이 일관되지 않을 수 있어요.

모호성을 피하기 위해 프로덕션 체인 secret에 정확히 한 가지 사용 가능한 자격 증명 타입을 구성할 것을 권장해요.

설정 (Settings)

delta 확장은 다음 설정을 추가해요.

기능 (Features)

delta 확장은 다음을 지원해요.

• 멀티스레드 스캔과 Parquet 메타데이터 읽기
• 데이터 스킵핑/필터 푸시다운
  • 파일 안의 행 그룹 스킵핑(Parquet 메타데이터 기반)
  • 전체 파일 스킵핑(Delta 파티션 정보 기반)
• 프로젝션 푸시다운
• 삭제 벡터가 있는 테이블 스캔
• 모든 원시 타입
• struct
• VARIANT 타입
• blind append (INSERT INTO)
• secret이 있는 클라우드 스토리지 (AWS S3, Azure, GCS)

지원 플랫폼

delta 확장은 현재 다음 플랫폼만을 지원해요.

• Linux AMD64 (x86_64 및 ARM64): linux_amd64 및 linux_arm64
• macOS Intel 및 Apple Silicon: osx_amd64 및 osx_arm64
• Windows AMD64: windows_amd64

[다른 DuckDB 플랫폼]({% link docs/current/extensions/extension_distribution.md %}#platforms)에 대한 지원은 진행 중이에요.

DuckDB와 함께 delta-rs 사용

이 예시에서는 delta-rs Python 바인딩으로 Delta 테이블을 만든 다음 DuckDB의 delta 확장으로 읽어요. Arrow zero-copy 통합으로 변경 데이터 피드(change data feed)를 읽는 것 같은 다른 읽기 작업을 DuckDB로 하는 방법도 보여줘요. 더 큰 데이터를 읽을 때는 Arrow Datasets를 사용해 이 작업을 지연(lazy)시킬 수도 있어요.

import deltalake as dl
import pyarrow as pa

# Create a delta table and read it with DuckDB Delta extension
dl.write_deltalake(
    "tmp/some_table",
    pa.table({
        "id": [1, 2, 3],
        "value": ["a", "b", "c"]
    })
)
with duckdb.connect() as conn:
    conn.execute("""
        INSTALL delta;
        LOAD delta;
    """)
    conn.sql("""
        SELECT * FROM delta_scan('tmp/some_table')
    """).show()

# Append some data and read the data change feed using the PyArrow integration
dl.write_deltalake(
    "tmp/some_table",
    pa.table({
        "id": [4, 5],
        "value": ["d", "e"]
    }),
    mode="append"
)
table = dl.DeltaTable("tmp/some_table").load_cdf(starting_version=1, ending_version=2)
with duckdb.connect() as conn:
    conn.register("t", table)
    conn.sql("SELECT * FROM t").show()

더 알아보기 (Learn more)