Lakeflow 파이프라인은 AUTO CDC 및 AUTO CDC FROM SNAPSHOT API를 사용하여 CDC(변경 데이터 캡처)를 간소화합니다. 이러한 API는 CDC 피드 또는 데이터베이스 스냅샷에서 SCD(느린 변경 차원) 유형 1 및 형식 2를 컴퓨팅하는 복잡성을 자동화합니다. 또한 AUTO CDC API는 두 개의 시간 차원에 걸친 변경 사항을 기록하는 이중 시간 추적(Beta)도 지원합니다. SCD 유형 1 및 형식 2에 대한 자세한 내용은 변경 데이터 캡처 및 스냅샷을 참조하세요. 이중 시간 추적에 대한 자세한 내용은 Bitemporal AUTO CDC를 참조하세요.
비고
API는 AUTO CDC API를 APPLY CHANGES 대체하고 구문이 동일합니다.
APPLY CHANGES API는 여전히 사용할 수 있지만, Databricks는 API 대신 AUTO CDC API를 사용하는 것을 권장합니다.
사용하는 API는 변경 데이터의 원본에 따라 달라집니다.
-
AUTO CDC: 원본 데이터베이스에 CDC 피드를 사용하도록 설정된 경우 사용합니다.AUTO CDC는 CDF(변경 데이터 피드)의 변경 내용을 처리합니다. 파이프라인 SQL 및 Python 인터페이스 모두에서 지원됩니다. -
AUTO CDC FROM SNAPSHOT: 원본 데이터베이스에서 CDC를 사용하도록 설정하지 않고 스냅샷만 사용할 수 있는 경우 사용합니다. 이 API는 스냅샷을 비교하여 변경 내용을 확인한 다음 처리합니다. 파이프라인 SQL 및 Python 인터페이스 모두에서 지원됩니다.
SCD 타입 1 타겟의 경우, 한 개의 AUTO CDC 플로우와 하나 이상의 AUTO CDC FROM SNAPSHOT 플로우를 결합할 수도 있습니다. 소스가 주기적인 신뢰할 수 있는 기준 스냅샷과 스냅샷 사이에 낮은 지연 시간의 CDC 피드를 제공할 때 이 패턴을 사용하세요.
두 API 모두 SCD Type 1 및 Type 2를 사용하여 테이블 업데이트를 지원합니다.
- SCD 유형 1을 사용하여 레코드를 직접 업데이트합니다. 레코드가 업데이트될 때, 이전 기록은 보존되지 않습니다.
- SCD 유형 2를 사용하여 모든 업데이트 또는 지정된 열 집합에 대한 업데이트에서 레코드 기록을 유지합니다.
단 AUTO CDC의 경우에만, SCD Type 2 이력을 확장하여 비즈니스 시간과 시스템 시간이라는 두 가지 시간 차원 전반의 변경 내용을 추적하는 이중 시간 저장소를 사용할 수도 있습니다. 바이템포럴은 베타 버전입니다.
Bitemporal AUTO CDC를 참조하세요.
AUTO CDC 또한 부분 업데이트를 지원합니다. 여기서 변경 레코드는 열의 하위 집합만 업데이트합니다.
부분 업데이트 적용을 참조하세요.
API는 AUTO CDC Apache Spark 선언적 파이프라인에서 지원되지 않습니다.
구문 및 기타 참조는 AUTO CDC INTO(파이프라인), create_auto_cdc_flow 및 create_auto_cdc_from_snapshot_flow 참조하세요.
비고
이 API를 사용해 소스 데이터 변경에 따라 파이프라인 내 테이블을 업데이트하세요. 델타 테이블에 대한 행 수준 변경 정보를 기록하고 쿼리하는 방법을 알아보려면 Azure Databricks 변경 데이터 피드 사용을 참조하세요.
요구 사항
CDC API를 사용하려면 파이프라인이 서버리스 Lakeflow 파이프라인 또는 Lakeflow 파이프라인 ProAdvanced에디션을 사용하도록 구성되어야 합니다.
AUTO CDC 작동 방식
CDC 처리를 AUTO CDC 수행하려면 스트리밍 테이블을 만든 다음, SQL에서는 AUTO CDC ... INTO 구문을, Python에서는 create_auto_cdc_flow() 함수를 사용하여 변경 피드에 대한 원본, 키 및 시퀀싱을 지정합니다. 시퀀싱 및 SCD 논리의 작동 방식에 대한 설명은 변경 데이터 캡처 및 스냅샷을 참조하세요.
AUTO CDC 예제를 참조하세요.
변경 피드가 있는 원본에서의 초기 하이드레이션을 위해 흐름과 함께 AUTO CDC을 사용하여 once를 실행하고, 그 후 변경 피드 처리를 계속하십시오.
AUTO CDC를 사용하여 외부 RDBMS 테이블 복제를 참조하세요.
구문 세부 정보를 보려면 AUTO CDC INTO (파이프라인) 또는 create_auto_cdc_flow를 참조하세요.
스냅샷의 자동 CDC 작동 방식
AUTO CDC FROM SNAPSHOT 는 순서대로 스냅샷을 비교하여 원본 데이터의 변경 내용을 결정합니다. 델타 테이블, 클라우드 스토리지 파일 또는 JDBC에서 직접 스냅샷을 읽을 수 있습니다. 파이프라인 SQL 및 Python 인터페이스 모두에서 지원됩니다.
CDC 처리 AUTO CDC FROM SNAPSHOT를 수행하려면 스트리밍 테이블을 생성하고 다음 흐름을 정의합니다:
- SQL에서는
CREATE STREAMING TABLE문의CREATE FLOW형식 또는AUTO CDC ... FROM SNAPSHOT의 내장FLOW AUTO CDC절을 사용하세요. 원본은 필수FROM SNAPSHOT (snapshot_query)절과 다음 스냅샷 버전을 선택하는 선택적WITH VERSION (version_query)절입니다. CREATE FLOW (파이프라인)을 참조하세요. - Python에서는
create_auto_cdc_from_snapshot_flow()함수를 사용해 스냅샷, 키, 기타 인자를 지정하세요. create_auto_cdc_from_snapshot_flow 참고하세요.
두 수집 패턴 및 각각을 사용하는 시기에 대한 자세한 내용은 스냅샷 처리 패턴을 참조하세요. AUTO CDC FROM SNAPSHOT 예제를 참조하세요.
스냅샷의 자동 CDC와 AUTO CDC를 결합하세요
스냅샷 버전과 CDC 시퀀싱 열은 하나의 순서 도메인을 형성합니다. 최신 스냅샷보다 새로운 CDC 이벤트는 스냅샷보다 우선 적용됩니다. 새로운 스냅샷이 이전 CDC 이벤트보다 우선하며, 스냅샷에 없는 키는 스냅샷 버전에서 삭제된 것으로 간주합니다. 최신 CDC 이벤트는 해당 키를 보존하거나 복원할 수 있습니다.
통합 목표는 다음 요건을 충족해야 합니다:
- 대상은 SCD 타입 1을 사용합니다.
- 타겟은 정확히 하나의
AUTO CDC FROM SNAPSHOT흐름과 하나 이상의AUTO CDC흐름을 가집니다. - 각
AUTO CDC흐름은 고유한 이름을 가집니다. - 모든 플로우는 같은 수의 키를 같은 순서로 사용합니다. 스냅샷 플로우 키명은
AUTO CDC키 이름과 대소문자를 구분하지 않고 비교됩니다. 여러AUTO CDC플로우는 동일한 키 이름과 케이싱을 사용해야 합니다. - 스냅샷 버전과 각 CDC 시퀀싱 열은 정확히 동일한 데이터 타입을 가집니다.
-
AUTO CDC FROM SNAPSHOT흐름에는 기대치가 정의되어 있지 않습니다. - 이 흐름들은
AUTO CDC또는 그 열 목록 변형을 사용하지 않습니다. - 파이프라인은 트리거 모드를 사용합니다.
- 기존
AUTO CDC타겟에AUTO CDC FROM SNAPSHOT플로우를 추가할 경우, 해당 타겟에는 예약된AUTO CDC시스템 컬럼과 이름이 충돌하는 사용자 컬럼이 포함되어서는 안 됩니다.
AUTO CDC 및 AUTO CDC FROM SNAPSHOT 흐름은 모두 Python 또는 SQL 파이프라인 인터페이스를 사용할 수 있습니다. SQL과 Python 플로우를 같은 타겟 내에서 혼합할 수 있습니다.
AUTO CDC 플로우에서 Python에서는 ONCE, SQL에서는 once=True을 사용하여 스냅샷을 한 번 백필하는 동안에도 AUTO CDC FROM SNAPSHOT 플로우는 이후 업데이트에서 새 이벤트를 계속 처리합니다. 대상의 전체 새로 고침은 일회성 스냅샷 흐름을 재실행하는 것입니다. 완전한 예시는 AUTO CDC SCD Type 1 테이블에 백필 추가를 참조하세요.
이 패턴은 SCD 유형 2 또는 이중 시간 대상을 지원하지 않습니다.
여러 열을 시퀀싱에 사용
여러 열(예를 들어 타임스탬프와 동점 판단을 위한 ID)에 따라 정렬하려면 STRUCT를 사용하여 결합합니다. API는 첫 번째 필드를 기준으로 정렬하고, 동률인 경우 두 번째 필드 등을 고려합니다.
SQL
SEQUENCE BY STRUCT(timestamp_col, id_col)
파이썬
sequence_by = struct("timestamp_col", "id_col")
AUTO CDC 예제
다음 예제에서는 변경 데이터 피드 원본을 사용하여 SCD Type 1 및 Type 2 처리를 보여 줍니다. 샘플 데이터는 새 사용자 레코드를 만들고, 사용자 레코드를 삭제하고, 사용자 레코드를 업데이트합니다. SCD Type 1 예제에서 마지막 UPDATE 작업은 늦게 도착하여 대상 테이블에서 삭제되어 순서가 다른 이벤트 처리를 보여 줍니다.
다음은 이러한 예제에서 사용되는 입력 레코드입니다. 이 데이터는 샘플 데이터 만들기 섹션에서 쿼리를 실행하여 생성됩니다.
| userId | 이름 | city | 수술 | 시퀀스번호 |
|---|---|---|---|---|
| 124 | 라울 | 오악사카 | INSERT | 1 |
| 123 | 이사벨 | 몬테레이 | INSERT | 1 |
| 125 | 메르세데스 | 티후아나 | INSERT | 2 |
| 126 | 백합 | 칸쿤 | INSERT | 2 |
| 123 | null | null | 삭제 | 6 |
| 125 | 메르세데스 | 과달라하라 | UPDATE | 6 |
| 125 | 메르세데스 | 멕시칼리 주 | UPDATE | 5 |
| 123 | 이사벨 | Chihuahua | UPDATE | 5 |
샘플 데이터 생성 쿼리에서 마지막 행의 주석 처리를 제거하면 sequenceNum=3 시점에 테이블을 비우라는 명령을 포함한 다음 레코드를 삽입합니다.
| userId | 이름 | city | 수술 | 시퀀스번호 |
|---|---|---|---|---|
| null | null | null | 잘라내기 | 3 |
비고
다음 예에는 DELETE 및 TRUNCATE 작업을 지정하는 옵션이 포함되어 있지만, 각각은 선택 사항입니다.
샘플 데이터 만들기
다음 문을 실행하여 샘플 데이터 세트를 만듭니다. 이 코드는 파이프라인 정의의 일부로 실행되지 않습니다. 변환 폴더가 아닌 파이프라인의 탐색 폴더에서 실행합니다.
CREATE SCHEMA IF NOT EXISTS main.cdc_tutorial;
CREATE TABLE main.cdc_tutorial.users_cdf
AS SELECT
col1 AS userId,
col2 AS name,
col3 AS city,
col4 AS operation,
col5 AS sequenceNum
FROM (
VALUES
-- Initial load.
(124, "Raul", "Oaxaca", "INSERT", 1),
(123, "Isabel", "Monterrey", "INSERT", 1),
-- New users.
(125, "Mercedes", "Tijuana", "INSERT", 2),
(126, "Lily", "Cancun", "INSERT", 2),
-- Isabel is removed from the system and Mercedes moved to Guadalajara.
(123, null, null, "DELETE", 6),
(125, "Mercedes", "Guadalajara", "UPDATE", 6),
-- This batch of updates arrived out of order. The batch at sequenceNum 6 is the final state.
(125, "Mercedes", "Mexicali", "UPDATE", 5),
(123, "Isabel", "Chihuahua", "UPDATE", 5)
-- Uncomment to test TRUNCATE.
-- ,(null, null, null, "TRUNCATE", 3)
);
SCD 유형 1 업데이트 처리
SCD 유형 1은 각 레코드의 최신 버전만 유지합니다. 다음 예제에서는 위에서 만든 변경 데이터 피드에서 읽고 스트리밍 테이블 대상에 변경 내용을 적용합니다. 이 코드를 실행하려면 파이프라인이 필요합니다. Lakeflow 파이프라인 편집기를 사용하여 ETL 파이프라인 개발 및 디버그를 참조하세요.
파이썬
from pyspark import pipelines as dp
from pyspark.sql.functions import col, expr
@dp.view
def users():
return spark.readStream.table("main.cdc_tutorial.users_cdf")
dp.create_streaming_table("users_current")
dp.create_auto_cdc_flow(
target = "users_current",
source = "users",
keys = ["userId"],
sequence_by = col("sequenceNum"),
apply_as_deletes = expr("operation = 'DELETE'"),
apply_as_truncates = expr("operation = 'TRUNCATE'"),
except_column_list = ["operation", "sequenceNum"],
stored_as_scd_type = 1
)
SQL
CREATE OR REFRESH STREAMING TABLE users_current;
CREATE FLOW apply_cdc AS AUTO CDC INTO
users_current
FROM
stream(main.cdc_tutorial.users_cdf)
KEYS
(userId)
APPLY AS DELETE WHEN
operation = "DELETE"
APPLY AS TRUNCATE WHEN
operation = "TRUNCATE"
SEQUENCE BY
sequenceNum
COLUMNS * EXCEPT
(operation, sequenceNum)
STORED AS
SCD TYPE 1;
SCD Type 1 예제를 실행한 후 대상 테이블에는 다음 레코드가 포함됩니다.
| userId | 이름 | city |
|---|---|---|
| 124 | 라울 | 오악사카 |
| 125 | 메르세데스 | 과달라하라 |
| 126 | 백합 | 칸쿤 |
사용자 123(Isabel)이 삭제되었으며 표시되지 않습니다. SCD 유형 1이 이전 값을 덮어쓰기 때문에 사용자 125(메르세데스)는 최신 도시(과달라하라)만 표시합니다. 나중에 UPDATE에서 업데이트가 도착했기 때문에 이전 sequenceNum=5의 sequenceNum=6가 삭제되었습니다.
예제를 실행할 때, 주석 처리되지 않은 레코드를 사용하면, 테이블은 TRUNCATE에서 초기화됩니다. 즉, 레코드 124 가 126 테이블에 없으며 최종 대상 테이블에는 다음 레코드만 포함됩니다.
| userId | 이름 | city |
|---|---|---|
| 125 | 메르세데스 | 과달라하라 |
SCD 유형 2 업데이트 처리
SCD 형식 2는 각 버전의 레코드에 대해 새 행을 만들고 각 버전이 활성화된 시기를 나타내는 열과 __START_AT 함께 __END_AT 변경 내용의 전체 기록을 유지합니다.
파이썬
from pyspark import pipelines as dp
from pyspark.sql.functions import col, expr
@dp.view
def users():
return spark.readStream.table("main.cdc_tutorial.users_cdf")
dp.create_streaming_table("users_history")
dp.create_auto_cdc_flow(
target = "users_history",
source = "users",
keys = ["userId"],
sequence_by = col("sequenceNum"),
apply_as_deletes = expr("operation = 'DELETE'"),
except_column_list = ["operation", "sequenceNum"],
stored_as_scd_type = "2"
)
SQL
CREATE OR REFRESH STREAMING TABLE users_history;
CREATE FLOW apply_cdc AS AUTO CDC INTO
users_history
FROM
stream(main.cdc_tutorial.users_cdf)
KEYS
(userId)
APPLY AS DELETE WHEN
operation = "DELETE"
SEQUENCE BY
sequenceNum
COLUMNS * EXCEPT
(operation, sequenceNum)
STORED AS
SCD TYPE 2;
SCD Type 2 예제를 실행한 후 대상 테이블에는 다음 레코드가 포함됩니다.
| userId | 이름 | city | __START_AT | __END_AT |
|---|---|---|---|---|
| 123 | 이사벨 | 몬테레이 | 1 | 5 |
| 123 | 이사벨 | Chihuahua | 5 | 6 |
| 124 | 라울 | 오악사카 | 1 | null |
| 125 | 메르세데스 | 티후아나 | 2 | 5 |
| 125 | 메르세데스 | 멕시칼리 주 | 5 | 6 |
| 125 | 메르세데스 | 과달라하라 | 6 | null |
| 126 | 백합 | 칸쿤 | 2 | null |
테이블은 전체 기록을 유지합니다. 사용자 123에는 두 가지 버전이 있습니다(삭제 시 시퀀스 6에서 종료됨). 사용자 125에는 도시 변경 내용을 보여 주는 세 가지 버전이 있습니다.
__END_AT = null 현재 활성 상태인 레코드입니다.
SCD 형식 2를 사용하여 열 하위 집합 추적
기본적으로 SCD Type 2는 열 값이 변경될 때마다 새 버전을 만듭니다. 추적할 열의 하위 집합을 지정하여 다른 열의 변경 내용이 새 기록 레코드를 생성하는 대신 현재 버전을 현재 상태로 업데이트하도록 할 수 있습니다.
다음 예제에서는 기록 추적에서 열을 제외합니다 city .
파이썬
from pyspark import pipelines as dp
from pyspark.sql.functions import col, expr
@dp.view
def users():
return spark.readStream.table("main.cdc_tutorial.users_cdf")
dp.create_streaming_table("users_history")
dp.create_auto_cdc_flow(
target = "users_history",
source = "users",
keys = ["userId"],
sequence_by = col("sequenceNum"),
apply_as_deletes = expr("operation = 'DELETE'"),
except_column_list = ["operation", "sequenceNum"],
stored_as_scd_type = "2",
track_history_except_column_list = ["city"]
)
SQL
CREATE OR REFRESH STREAMING TABLE users_history;
CREATE FLOW apply_cdc AS AUTO CDC INTO
users_history
FROM
stream(main.cdc_tutorial.users_cdf)
KEYS
(userId)
APPLY AS DELETE WHEN
operation = "DELETE"
SEQUENCE BY
sequenceNum
COLUMNS * EXCEPT
(operation, sequenceNum)
STORED AS
SCD TYPE 2
TRACK HISTORY ON * EXCEPT
(city)
city 변경 내용은 추적되지 않으므로 도시 업데이트는 새 버전을 만드는 대신 현재 행을 덮어씁니다. 대상 테이블에는 다음 레코드가 포함됩니다.
| userId | 이름 | city | __START_AT | __END_AT |
|---|---|---|---|---|
| 123 | 이사벨 | Chihuahua | 1 | 6 |
| 124 | 라울 | 오악사카 | 1 | null |
| 125 | 메르세데스 | 과달라하라 | 2 | null |
| 126 | 백합 | 칸쿤 | 2 | null |
AUTO CDC FROM SNAPSHOT 예제
다음 섹션에서는 SCD 유형 1 또는 형식 2 대상 테이블로 스냅샷을 처리하는 데 사용하는 AUTO CDC FROM SNAPSHOT 예제를 제공합니다. 이 API를 사용하는 시기에 대한 배경 정보는 변경 데이터 캡처 및 스냅샷을 참조하세요.
이 섹션의 예시로는 Python과 SQL 구현이 포함됩니다. 완전한 SQL 문법은 CREATE FLOW (pipelines)를 참조하세요.
예시: 주기적 스냅샷 처리
스냅샷이 규칙적이고 순서대로 도착할 때 이 방식을 사용하세요. Python 인터페이스는 버전 관리를 위해 파이프라인 업데이트 순서를 사용합니다. SQL의 경우, 예시는 새로운 스냅샷이 언제 사용 가능해지는지 식별하기 위해 별도의 버전 테이블을 사용합니다.
델타 테이블, 클라우드 스토리지 파일 및 JDBC 연결을 비롯한 여러 원본 형식에서 스냅샷을 읽을 수 있습니다.
1단계: 샘플 데이터 만들기
스냅샷 데이터가 포함된 테이블을 만듭니다. 파이프라인의 explorations 폴더에서 Notebook 또는 Databricks SQL을 사용하여 다음 코드를 실행합니다.
CREATE SCHEMA IF NOT EXISTS main.cdc_tutorial;
CREATE TABLE main.cdc_tutorial.snapshot (
userId INT,
city STRING
);
CREATE TABLE main.cdc_tutorial.snapshot_version (
version BIGINT
);
INSERT INTO main.cdc_tutorial.snapshot VALUES
(1, 'Oaxaca'),
(2, 'Monterrey'),
(3, 'Tijuana');
INSERT INTO main.cdc_tutorial.snapshot_version VALUES (0);
2단계: 스냅샷에서 자동 CDC 실행
이 단계에서 코드를 실행하려면 파이프라인이 필요합니다. Lakeflow 파이프라인 편집기를 사용하여 ETL 파이프라인 개발 및 디버그를 참조하세요.
Python의 경우, 스냅샷 뷰의 소스 유형을 선택하세요(샘플 생성 코드가 Delta 테이블을 생성합니다):
옵션 A: 델타 테이블에서 읽기
from pyspark import pipelines as dp
@dp.view(name="source")
def source():
return spark.read.table("main.cdc_tutorial.snapshot")
옵션 B: 클라우드 스토리지에서 읽기
from pyspark import pipelines as dp
@dp.view(name="source")
def source():
return spark.read.format("csv").option("header", True).load("<snapshot-path>")
옵션 C: JDBC에서 읽기(클래식 컴퓨팅에만 해당)
from pyspark import pipelines as dp
@dp.view(name="source")
def source():
return (spark.read
.format("jdbc")
.option("url", "<jdbc-url>")
.option("dbtable", "<table-name>")
.option("user", "<username>")
.option("password", "<password>")
.load()
)
그 다음 대상 테이블과 흐름을 추가하세요. SQL 구현은 새 스냅샷을 사용할 수 있을 때를 식별하기 위해 snapshot_version 테이블을 사용합니다:
파이썬
dp.create_streaming_table("target")
dp.create_auto_cdc_from_snapshot_flow(
target = "target",
source = "source",
keys = ["userId"],
stored_as_scd_type = 2
)
SQL
CREATE OR REFRESH STREAMING TABLE target;
CREATE FLOW target_snapshot_flow AS
AUTO CDC INTO target
FROM SNAPSHOT (
SELECT userId, city
FROM main.cdc_tutorial.snapshot
)
WITH VERSION (
SELECT version
FROM main.cdc_tutorial.snapshot_version
WHERE (
NOT EXISTS (SELECT 1 FROM last_snapshot_version())
OR version > (SELECT version FROM last_snapshot_version())
)
)
KEYS (userId)
STORED AS SCD TYPE 2;
첫 번째 파이프라인 실행 후 모든 레코드가 활성 행으로 삽입됩니다.
| userId | city | __START_AT | __END_AT |
|---|---|---|---|
| 1 | 오악사카 | 0 | null |
| 2 | 몬테레이 | 0 | null |
| 3 | 티후아나 | 0 | null |
비고
대신 SCD 타입 1을 사용하고 현재 상태만 유지하려면, Python에서는 STORED AS SCD TYPE 1, SQL에서는 stored_as_scd_type=1로 설정하세요. 이 경우 대상 테이블에는 __START_AT 열과 __END_AT 열이 포함되지 않습니다.
3단계: 새 스냅샷 시뮬레이션 및 다시 실행
소스 테이블을 업데이트하여 새로운 스냅샷이 도착하는 것을 시뮬레이션하세요 (파이프라인 폴더의 explorations 노트북이나 SQL 파일에서 이 코드를 실행하세요):
TRUNCATE TABLE main.cdc_tutorial.snapshot;
INSERT INTO main.cdc_tutorial.snapshot VALUES
(2, 'Carmel'),
(3, 'Los Angeles'),
(4, 'Death Valley'),
(6, 'Kings Canyon');
UPDATE main.cdc_tutorial.snapshot_version SET version = 1;
파이프라인 다시 실행
AUTO CDC FROM SNAPSHOT 는 새 스냅샷을 이전 스냅샷과 비교하고 사용자 1이 삭제되고, 사용자 2와 3이 업데이트되었으며, 사용자 4와 6이 삽입되었음을 감지합니다. 이렇게 하면 변경 피드가 생성되고 출력 테이블을 만드는 데 사용됩니다 AUTO CDC .
SCD 형식 2를 사용하여 두 번째 실행 후 대상 테이블에는 다음 레코드가 포함됩니다.
| userId | city | __START_AT | __END_AT |
|---|---|---|---|
| 1 | 오악사카 | 0 | 1 |
| 2 | 몬테레이 | 0 | 1 |
| 2 | 카멜 | 1 | null |
| 3 | 티후아나 | 0 | 1 |
| 3 | 로스앤젤레스 | 1 | null |
| 4 | 데스 밸리 | 1 | null |
| 6 | 킹스 캐년 | 1 | null |
사용자 1이 종료되었습니다(삭제됨). 사용자 2와 3에는 각각 도시 변경 내용을 표시하는 두 가지 버전이 있습니다. 사용자 4와 6이 새로 삽입되었습니다.
SCD 형식 1을 사용하여 두 번째 실행 후 대상 테이블에는 현재 상태만 표시됩니다.
| userId | city |
|---|---|
| 2 | 카멜 |
| 3 | 로스앤젤레스 |
| 4 | 데스 밸리 |
| 6 | 킹스 캐년 |
예: 버전 함수를 사용하여 스냅샷 처리
스냅샷 순서를 명시적으로 제어해야 하는 경우 이 방법을 사용합니다. 예를 들어 여러 스냅샷이 동시에 도착하거나 스냅샷이 순서를 벗어나는 경우 이 방법을 사용합니다. 다음 스냅샷과 버전 번호를 선택하는 방법을 정의합니다. API는 스냅샷을 오름차순으로 처리합니다.
- 여러 개의 스냅샷이 저장소에 있을 경우, 예제 셀렉터는 사용 가능한 스냅샷을 버전 순서대로 처리합니다.
- 만약 스냅샷이 순서와 다르게 도착하면(예를 들어
snapshot_3가snapshot_4뒤에 도착하는 경우), 예제 셀렉터는 해당 스냅샷의 버전이 마지막으로 커밋된 버전보다 크지 않기 때문에 이를 제외합니다. - 새로운 스냅샷이 없으면 버전 함수나 쿼리는 결과를 반환하지 않으며 처리가 이루어지지 않습니다.
1단계: 스냅샷 파일 준비
스냅샷 데이터가 포함된 CSV 파일을 만들고 볼륨 또는 클라우드 스토리지 위치에 추가합니다. 파일 이름을 시간순으로 지정합니다(예: snapshot_1.csv, snapshot_2.csv).
각 파일에는 userId 및 city에 대한 열이 포함되어야 합니다. 다음은 그 예입니다.
snapshot_1.csv:
| userId | city |
|---|---|
| 1 | 오악사카 |
| 2 | 몬테레이 |
| 3 | 티후아나 |
snapshot_2.csv:
| userId | city |
|---|---|
| 2 | 카멜 |
| 3 | 로스앤젤레스 |
| 4 | 데스 밸리 |
2단계: 버전 함수를 사용하여 SNAPSHOT에서 AUTO CDC 실행
파이프라인에서 transformations 폴더에 새로운 Python 또는 SQL 파일을 만들고 다음 코드를 추가하세요. 그런 다음, 파이프라인을 실행합니다.
Lakeflow 파이프라인 편집기를 사용하여 ETL 파이프라인 개발 및 디버그를 참조하세요.
파이썬
from pyspark import pipelines as dp
from typing import Optional, Tuple
from pyspark.sql import DataFrame
def next_snapshot_and_version(latest_snapshot_version: Optional[int]) -> Optional[Tuple[DataFrame, int]]:
snapshot_dir = "/Volumes/main/cdc_tutorial/snapshots/" # or the location you created the sample data
files = dbutils.fs.ls(snapshot_dir)
snapshot_files = [f.name for f in files if f.name.startswith("snapshot_") and f.name.endswith(".csv")]
snapshot_versions = []
for filename in snapshot_files:
try:
version = int(filename.replace("snapshot_", "").replace(".csv", ""))
snapshot_versions.append(version)
except ValueError:
continue
snapshot_versions.sort()
if latest_snapshot_version is None:
if snapshot_versions:
next_version = snapshot_versions[0]
else:
return None
else:
next_versions = [v for v in snapshot_versions if v > latest_snapshot_version]
if next_versions:
next_version = next_versions[0]
else:
return None
snapshot_path = f"{snapshot_dir}snapshot_{next_version}.csv"
df = spark.read.format("csv").option("header", True).load(snapshot_path)
return (df, next_version)
dp.create_streaming_table("main.cdc_tutorial.target_versioned")
dp.create_auto_cdc_from_snapshot_flow(
target = "main.cdc_tutorial.target_versioned",
source = next_snapshot_and_version,
keys = ["userId"],
stored_as_scd_type = 2
)
SQL
CREATE OR REFRESH STREAMING TABLE main.cdc_tutorial.target_versioned;
CREATE FLOW target_versioned_snapshot_flow AS
AUTO CDC INTO main.cdc_tutorial.target_versioned
FROM SNAPSHOT (
SELECT CAST(userId AS INT) AS userId, city
FROM read_files(
'/Volumes/main/cdc_tutorial/snapshots/',
format => 'csv',
header => true
)
WHERE _metadata.file_name = CONCAT(
'snapshot_',
CAST((SELECT version FROM current_snapshot_version()) AS STRING),
'.csv'
)
)
WITH VERSION (
WITH snapshot_files AS (
SELECT
CAST(REGEXP_EXTRACT(path, 'snapshot_([0-9]+)[.]csv$', 1) AS BIGINT) AS version
FROM list_files('/Volumes/main/cdc_tutorial/snapshots/')
WHERE path RLIKE 'snapshot_[0-9]+[.]csv$'
)
SELECT version
FROM snapshot_files
WHERE (
NOT EXISTS (SELECT 1 FROM last_snapshot_version())
OR version > (SELECT version FROM last_snapshot_version())
)
ORDER BY version
LIMIT 1
)
KEYS (userId)
STORED AS SCD TYPE 2;
비고
SCD 타입 1을 사용하려면 Python에서는 STORED AS SCD TYPE 1로 설정하거나 SQL에서는 stored_as_scd_type=1로 설정하세요.
처리 snapshot_1.csv후 대상 테이블에는 다음 레코드가 포함됩니다.
| userId | city | __START_AT | __END_AT |
|---|---|---|---|
| 1 | 오악사카 | 1 | null |
| 2 | 몬테레이 | 1 | null |
| 3 | 티후아나 | 1 | null |
처리 snapshot_2.csv후 대상 테이블에는 다음 레코드가 포함됩니다.
| userId | city | __START_AT | __END_AT |
|---|---|---|---|
| 1 | 오악사카 | 1 | 2 |
| 2 | 몬테레이 | 1 | 2 |
| 2 | 카멜 | 2 | null |
| 3 | 티후아나 | 1 | 2 |
| 3 | 로스앤젤레스 | 2 | null |
| 4 | 데스 밸리 | 2 | null |
비고
SCD 유형 1의 경우 테이블은 가장 최근의 스냅샷과 똑같습니다. 차이점은 다운스트림 쿼리가 변경 피드를 사용하여 변경된 레코드만 처리할 수 있다는 점입니다.
3단계: 새 스냅샷 추가
수정된 데이터(예: 변경된 도시 값, 새 행 또는 제거된 행)를 사용하여 스토리지 위치에 새 CSV 파일을 추가합니다. 그런 다음 파이프라인을 다시 실행하여 새 스냅샷을 처리합니다.
제한점
- 시퀀싱 열은 정렬 가능한 데이터 형식이어야 합니다.
NULL시퀀싱 값은 지원되지 않습니다. - AUTO CDC 프로세스의 대상에서 데이터를 스트리밍하려면 변경 피드에서 읽습니다. 자세한 내용은 AUTO CDC 대상 테이블에서 변경 데이터 피드 읽기를 참조하세요.
추가 리소스
- 데이터 캡처 및 스냅샷 변경: CDC 개념, 스냅샷 및 SCD 형식에 대해 알아봅니다.
- 다음
AUTO CDC에 대해 알아봅니다: 워크플로우를 통해 초기 하이드레이션을 수행한 다음 변경 사항을 계속 처리합니다. - 파이프라인을 통한 과거 데이터 백필: append-only와 AUTO CDC SCD Type 1 타겟을 백필하는 방법을 배우세요.
- 고급 AUTO CDC 항목: AUTO CDC 대상에 대한 변경 작업, 변경 데이터 피드 읽기 및 메트릭 처리에 대해 알아봅니다.
- 파이프라인을 되감고 재생하기: AUTO CDC 목표를 이전 시점으로 복원하고 변경 사항을 재처리하는 방법을 배우세요.
- 자습서: 변경 데이터 캡처를 사용하여 ETL 파이프라인 빌드