현재 스냅샷을 이력 데이터셋으로 누적하기
매일이나 매시간 SNAPSHOT으로 교체되는 원천의 이전 상태를 보존하려면, 가능하면 수집 단계에서 가져온 날짜/시각을 열로 추가하고 APPEND로 적재합니다. 이렇게 수집할 수 없을 때는 현재 스냅샷을 별도 증분 출력에 쌓을 수 있습니다. 다만 실행 사이에 놓친 스냅샷을 이 방법으로 복구할 수는 없습니다.
현재 입력 전체에 수집 날짜 붙이기
입력을 snapshot_inputs로 선언하면 기본 읽기에서도 현재 전체 뷰를 읽습니다. 출력은 첫 실행에서 replace, 이후 증분 실행에서 modify를 사용하므로 새 스냅샷의 행이 앞선 이력 뒤에 추가됩니다.
다음 예제에서는 입력에 date라는 열이 없다고 가정합니다. 이미 같은 이름의 업무 열이 있다면 수집 날짜에 다른 열 이름을 사용합니다. 자산 경로는 가상의 예시입니다.
Polars
from datetime import date
import polars as pl
from transforms.api import Input, Output, incremental, transform
@incremental(snapshot_inputs=["input_data"])
@transform.using(
input_data=Input("/Example Project/Inputs/current_snapshot"),
history=Output("/Example Project/Outputs/history_polars"),
)
def compute(input_data, history):
rows = input_data.polars().with_columns(pl.lit(date.today()).alias("date"))
history.write_table(rows)
DuckDB
from transforms.api import Input, Output, incremental, transform
@incremental(snapshot_inputs=["input_data"])
@transform.using(
input_data=Input("/Example Project/Inputs/current_snapshot"),
history=Output("/Example Project/Outputs/history_duckdb"),
)
def compute(ctx, input_data, history):
result = ctx.duckdb().conn.sql("SELECT *, CURRENT_DATE AS date FROM input_data")
history.write_table(result)
Pandas
import pandas as pd
from transforms.api import Input, Output, incremental, transform
@incremental(snapshot_inputs=["input_data"])
@transform.using(
input_data=Input("/Example Project/Inputs/current_snapshot"),
history=Output("/Example Project/Outputs/history_pandas"),
)
def compute(input_data, history):
rows = input_data.pandas()
rows["date"] = pd.Timestamp.now().date()
history.write_table(rows)
PySpark
from pyspark.sql import functions as F
from transforms.api import Input, Output, incremental, transform
@incremental(snapshot_inputs=["input_data"])
@transform.spark.using(
input_data=Input("/Example Project/Inputs/current_snapshot"),
history=Output("/Example Project/Outputs/history_spark"),
)
def compute(input_data, history):
history.write_dataframe(input_data.dataframe().withColumn("date", F.current_date()))
하루에 여러 번 입력이 바뀌면 날짜만으로 상태를 구분할 수 없습니다. Polars에서는 from datetime import datetime으로 가져온 datetime.now(), DuckDB에서는 CURRENT_TIMESTAMP, Pandas에서는 .date()를 생략한 pd.Timestamp.now(), Spark에서는 F.current_timestamp()를 사용합니다. 업무상 시간대 기준도 정합니다. 이 열은 실행 중 기록한 수집 시각이며 원천의 실제 사건 발생 시각과 구분합니다.
현재 입력의 표를 그대로 쓰므로 이전 출력을 읽기 위한 별도 빈 schema는 필요하지 않습니다. 스냅샷 참조 입력의 added도 현재 전체를 읽는다는 특수 읽기 규칙이 적용됩니다.
누적을 관찰하는 순서
- 첫 빌드를 수행합니다. 처음 만든 출력은 비증분으로 실행되어 현재 스냅샷 한 개만 가지는 것이 정상입니다.
- 재수집이나 재동기화로 입력에 다른
SNAPSHOT트랜잭션을 만든 뒤 다시 빌드합니다. - 날짜나 시각 열을 기준으로 두 상태의 행이 모두 있는지 확인합니다.
- Dataset Preview, Data Lineage 또는 Builds에서 빌드 보고서를 열고 Gantt 차트 아래의 해당 작업을 선택합니다. Spark 작업에서는
Spark details의Snapshot/Incremental탭에서 입력 범위와 증분 실행 여부를 확인합니다. 경량 변환에서는 계산 함수의ctx인자를 통해ctx.is_incremental값을 로그로 확인할 수 있습니다.
새 입력 트랜잭션이나 로직 변경이 없는 일반 빌드는 최신 상태로 판단되어 계산을 생략할 수 있습니다. 위 코드에는 동일 상태의 중복 기록을 제거하는 논리가 없으므로 강제 실행을 반복하는 운영 방식은 별도로 판단합니다.
Code Repositories Preview는 항상 비증분으로 실행하여 현재 스냅샷으로 교체하므로 이력 누적을 확인하는 도구로 사용하지 않습니다. VS Code preview의 증분 실행이나 실제 수동/예약 빌드 결과를 확인합니다.
서로 다른 입력 트랜잭션을 처리한 두 빌드 뒤에도 최신 상태만 남았다면 두 번째 빌드가 스냅샷으로 실행되었는지 조사합니다. semantic version 변경, 전체 교체 입력의 snapshot_inputs 누락, 다른 변환이 만든 출력의 마지막 커밋을 확인하고 드라이버 로그의 transforms.api._incremental: Not running incrementally 경고를 살펴봅니다.
보존과 스키마 변경의 위험
빌드 실패나 실행 간격 때문에 중간 입력 스냅샷을 놓치면 이 변환은 놓친 상태를 다시 읽을 수 없습니다. 모든 상태의 보존이 필요하면 데이터 소유자와 수집 방식을 APPEND로 바꾸는 방안을 마련합니다.
입력의 열 추가, 열 개수 변경이나 integer에서 decimal로의 자료형 변경은 기존 출력과 새 배치의 schema 호환성을 확인해야 합니다. 쓰기 경로가 허용하지 않는 열 구성이나 자료형 변경은 실패합니다. 새 열 추가를 허용하는 경로에서도 과거 행에는 그 값이 없으므로 이력 전체가 자동으로 채워지지 않습니다.
입력 데이터셋 자체를 교체할 때도 같은 schema라는 이유만으로 기존 이력과 연속성을 가정하지 않습니다. 스냅샷 재실행이 선택되면 현재 입력으로 출력 전체를 바꾸므로 앞서 누적한 이력이 사라질 수 있습니다. semantic version 변경이나 v2 의미론 전환처럼 전체 재실행을 유발하는 변경 전에 보존할 이력을 확인합니다.
작은 파일 누적
반복 추가는 작은 파일을 많이 만들 수 있습니다. 파일마다 읽기 오버헤드가 있어 하위 배치/대화형 계산 시간이 증가하고, 압축이 파일 단위로 이루어져 저장 공간도 늘어날 수 있습니다. 파일 수에 따른 병합은 이전 출력과 새 배치를 합쳐 다시 쓰는 네 엔진의 예제입니다.
이력 데이터를 재작성하는 작업은 기존 이력 전체를 포함해야 합니다. 현재 원천 스냅샷 하나로 이력 출력을 다시 만드는 작업과 혼동하지 않습니다.