이 함수는 create_auto_cdc_from_snapshot_flow Lakeflow 파이프라인 CDC(변경 데이터 캡처) 기능을 사용하여 데이터베이스 스냅샷에서 원본 데이터를 처리하는 흐름을 만듭니다.
스냅샷의 자동 CDC 작동 방식을 참조하세요.
비고
이 함수는 이전 함수 apply_changes_from_snapshot()를 대체합니다. 두 함수의 서명은 동일합니다. Databricks는 새 이름을 사용하도록 업데이트하는 것이 좋습니다.
중요합니다
이 작업에 대한 대상 스트리밍 테이블이 있어야 합니다. 필요한 대상 테이블을 만들려면 create_streaming_table() 함수를 사용할 수 있습니다.
Syntax
from pyspark import pipelines as dp
dp.create_auto_cdc_from_snapshot_flow(
target = "<target-table>",
source = Any,
keys = ["key1", "key2", "keyN"],
stored_as_scd_type = "1",
track_history_column_list = None,
track_history_except_column_list = None,
once = False
)
비고
처리를 위해 AUTO CDC FROM SNAPSHOT 동일한 키를 가진 일치하는 레코드가 대상에 없는 경우 새 행을 삽입하는 것이 기본 동작입니다. 일치하는 레코드가 있는 경우 행의 값이 변경된 경우에만 업데이트됩니다. 대상에 키가 있지만 더 이상 원본에 없는 행이 삭제됩니다.
스냅샷을 사용한 CDC 처리에 대한 자세한 내용은 AUTO CDC API: 파이프라인을 사용하여 변경 데이터 캡처 간소화를 참조하세요. 함수 사용 create_auto_cdc_from_snapshot_flow() 예시는 주기적 스냅샷 인제스팅, 히스토리 스냅샷 인제스텐션, SCD 타입 1 백필 예제를 참고하세요.
매개 변수
| 매개 변수 | 유형 | Description |
|---|---|---|
target |
str |
필수 사항입니다. 업데이트할 테이블의 이름입니다. 함수를 실행하기 전에 create_streaming_table() 함수를 사용하여 대상 테이블을 만들 수 있습니다 create_auto_cdc_from_snapshot_flow() . |
source |
str 또는 lambda function |
필수 사항입니다. 주기적으로 스냅샷할 테이블 또는 뷰의 이름 또는 처리할 스냅샷 DataFrame 및 스냅샷 버전을 반환하는 Python 람다 함수입니다.
인수 구현을 source 참조하세요. |
keys |
list |
필수 사항입니다. 원본 데이터의 행을 고유하게 식별하는 열 또는 열의 조합입니다. 이는 대상 테이블의 특정 레코드에 적용되는 CDC 이벤트를 식별하는 데 사용됩니다. 다음 중 하나를 지정할 수 있습니다.
|
stored_as_scd_type |
str 또는 int |
레코드를 SCD 형식 1 또는 SCD 형식 2로 저장할지 여부입니다.
1 SCD 유형 1 또는 2 SCD 유형 2로 설정합니다. 기본값은 SCD 형식 1입니다. |
track_history_column_list 또는 track_history_except_column_list |
list |
대상 테이블의 기록에 대해 추적할 출력 열의 하위 집합입니다. 추적할 열의 전체 목록을 지정하는 데 사용합니다 track_history_column_list . 추적에서 제외할 열을 지정하는 데 사용합니다 track_history_except_column_list . 값을 문자열 목록 또는 Spark SQL col() 함수로 선언할 수 있습니다.
함수에 대한 col() 인수에는 한정자를 포함할 수 없습니다. 예를 들어 사용할 col(userId)수 있지만 사용할 col(source.userId)수는 없습니다. 함수에 track_history_column_list 또는 track_history_except_column_list 인수가 전달되지 않으면 기본값으로 대상 테이블의 모든 열이 포함됩니다. |
once |
bool |
스냅샷 플로우를 한 번만 실행할지 여부. 이 있을 때 True, 성공적으로 커밋된 후 이후 추가 업데이트에서 플로우가 건너뛸 수 있습니다. 대상의 전체 새로고침이 흐름을 재실행합니다. 이 옵션을 사용하면 일회성 스냅샷 로드나 백필을 할 수 있습니다. 기본값은 False입니다. |
Notes
트리거 파이프라인의 SCD 타입 1 타겟의 경우, 한 create_auto_cdc_from_snapshot_flow() 플로우가 Python 또는 SQL로 정의된 하나 이상의 AUTO CDC 플로우와 타겟을 공유할 수 있습니다. 적용되는 요구 사항은 다음과 같습니다.
- 각 흐름마다
AUTO CDC고유한 이름을 붙이세요. - 모든 흐름에 같은 수의 키를 같은 순서로 사용하세요. 스냅샷 플로우 키명은 키명과
AUTO CDC대소문자 구분에 따라 비교됩니다. 여러 플로우는AUTO CDC동일한 키명과 케이싱을 사용해야 합니다. - 스냅샷 버전과 각
AUTO CDC플로우의 시퀀싱 열에 대해 정확히 동일한 데이터 타입을 사용하세요. - 흐름에 대한
AUTO CDC FROM SNAPSHOT기대치를 정의하지 마세요. -
IGNORE NULL UPDATES는 사용하지 마세요. Python에서는 , 또는ignore_null_updates_except_column_listignore_null_updates_column_list플로우에 대해create_auto_cdc_flow()설정ignore_null_updates하지 마세요. - 기존
AUTO CDC FROM SNAPSHOT타겟에 플로우를AUTO CDC추가할 경우, 해당 타겟에는 예약AUTO CDC된 시스템 컬럼과 이름이 충돌하는 사용자 컬럼이 포함되어서는 안 됩니다. - SCD 타입 2 또는 양측두엽 표적에는 이 패턴을 사용하지 마십시오.
예시로는 AUTO CDC SCD Type 1 테이블에 백필 추가를 참조하세요.
인수 구현 source
create_auto_cdc_from_snapshot_flow() 함수는 source 인수를 포함합니다. 기록 스냅샷을 처리하기 위해 `source` 인수는 처리할 스냅샷 데이터가 포함된 Python DataFrame과 스냅샷 버전이라는 두 값을 `create_auto_cdc_from_snapshot_flow()` 함수에 반환하는 Python 람다 함수여야 합니다.
함수의 첫 호출은 스냅샷 데이터프레임과 버전을 반환해야 합니다. 첫 번째 호출이 반환 None되면 파이프라인 업데이트는 실패합니다. 최소 하나의 스냅샷이 처리된 후에는 추가 스냅샷이 없음을 나타내기 위해 return None 을 표시합니다.
다음은 람다 함수의 서명입니다.
lambda Any => Optional[(DataFrame, Any)]
- 람다 함수에 대한 인수는 가장 최근에 처리된 스냅샷 버전입니다.
- 람다 함수의 반환 값은 두 값의 튜플입니다
None. 튜플의 첫 번째 값은 처리할 스냅샷을 포함하는 DataFrame입니다. 튜플의 두 번째 값은 스냅샷의 논리적 순서를 나타내는 스냅샷 버전입니다.
람다 함수를 구현하고 호출하는 예제:
def next_snapshot_and_version(latest_snapshot_version: Optional[int]) -> Tuple[DataFrame, Optional[int]]:
if latest_snapshot_version is None:
return (spark.read.load("filename.csv"), 1)
else:
return None
create_auto_cdc_from_snapshot_flow(
# ...
source = next_snapshot_and_version,
# ...
)
Lakeflow 파이프라인 런타임은 함수를 포함하는 파이프라인이 트리거될 때마다 다음 단계를 수행합니다 create_auto_cdc_from_snapshot_flow() .
- 함수를
next_snapshot_and_version실행하여 다음 스냅샷 DataFrame 및 해당 스냅샷 버전을 로드합니다. - 첫 번째 호출에서 DataFrame이 반환되지 않으면 업데이트가 실패합니다. 이후 호출에서 DataFrame이 반환되지 않으면 실행이 종료되고 파이프라인 업데이트는 완료로 표시됩니다.
- 새 스냅샷의 변경 내용을 검색하고 증분 방식으로 대상 테이블에 적용합니다.
- 1단계로 돌아와서 다음 스냅샷 및 해당 버전을 로드합니다.