Funnel 배치 파이프라인

Funnel 배치 파이프라인은 입력 데이터와 사용자 편집을 OSv2 객체 데이터베이스에 반영하는 내부 빌드 작업들의 집합입니다. 인덱싱 파일을 만든 뒤 저장소가 조회할 준비를 마쳐야 앱에서 새 상태를 사용할 수 있습니다.

네 가지 처리 단계

중간 변경 로그, 병합 데이터셋과 인덱스 데이터셋은 Funnel이 소유하고 관리하며 사용자가 직접 접근할 수 없습니다.

Live와 Replacement

Live 파이프라인은 입력 데이터소스가 갱신될 때 운영 객체 데이터를 갱신합니다. 새 원천 데이터가 없어도 사용자 편집을 감지하면 6시간마다 실행하여 내부 데이터셋에 편집을 영구 보존합니다. 사용자 편집 자체는 인덱스에 즉시 적용하므로 이 6시간은 앱에서 편집을 보기까지 기다리는 시간이 아닙니다.

속성 추가, 속성 자료형 변경이나 입력 데이터소스 교체처럼 스키마가 달라지면 Replacement 파이프라인을 준비합니다. 기존 Live는 평소 주기로 계속 실행하고 대체 파이프라인을 백그라운드에서 만듭니다. 대체 파이프라인이 처음으로 성공하면 기존 파이프라인을 교체하여 새 스키마와 데이터를 제공합니다. 성능상의 판단으로 자동 대체 파이프라인을 만들기도 합니다.

기본 증분 인덱싱

OSv2는 모든 객체 유형에 기본으로 증분 인덱싱을 사용합니다. 새 트랜잭션의 차이를 계산하고 변경된 데이터만 인덱싱합니다. 예를 들어 100행 중 10행이 바뀌면 입력 트랜잭션 종류와 무관하게 내부 변경 로그에 그 10행을 담은 새 APPEND 트랜잭션을 만듭니다.

증분 데이터셋의 키와 삭제

같은 기본 키가 여러 트랜잭션에 있으면 가장 나중 트랜잭션의 행을 사용합니다. 한 트랜잭션 안에 같은 기본 키가 중복되어서는 안 됩니다. 이 most recent transaction wins는 사용자 편집과 원천 갱신의 충돌 해결과 다른 규칙입니다.

is_deleted라는 이름만으로 열을 삭제 표시로 처리하지 않습니다. 삭제 열은 데이터소스 스키마의 레거시 OSv1 변경 로그 메타데이터에 지정되어 있을 때만 해석합니다. 인식되는 삭제 메타데이터를 사용하는 APPEND 경로에서는 삭제 표시 행을 없애지 말고 함께 전달해야 이전에 들어온 객체도 삭제할 수 있습니다. CDC 메타데이터는 기본 키, 순서 열과 삭제 열의 역할을 설명합니다.

다음 PySpark 함수는 입력의 기본 키 primary_key, 순서 ordering과 Boolean 삭제 표시 is_deleted를 사용합니다. 키와 순서 및 삭제 표시는 null이 없어야 하고, ordering은 키별 마지막 행을 유일하게 결정해야 합니다. __latest_rank는 함수가 쓰는 임시 열이므로 입력에 두지 않습니다.

from pyspark.sql import DataFrame, functions as F
from pyspark.sql.window import Window


def latest_changelog_rows(rows: DataFrame) -> DataFrame:
    ordering_window = Window.partitionBy("primary_key").orderBy(
        F.col("ordering").desc()
    )
    return (
        rows.withColumn("__latest_rank", F.row_number().over(ordering_window))
        .filter(F.col("__latest_rank") == 1)
        .drop("__latest_rank")
    )


def current_changelog_snapshot(full_history: DataFrame) -> DataFrame:
    return latest_changelog_rows(full_history).filter(~F.col("is_deleted"))

latest_changelog_rows에 한 출력 트랜잭션의 변경들을 넘기면 키별로 한 행만 남기되 삭제 표시를 보존합니다. 결과를 APPEND로 반영할 때는 앞서 설명한 삭제 메타데이터를 별도로 유지해야 합니다.

current_changelog_snapshot은 전체 변경 이력을 입력받아 현재 살아 있는 행을 반환합니다. 이 결과는 전체 상태를 교체하는 SNAPSHOT으로 써야 합니다. 새 증분 행만 넘기거나 결과를 APPEND로 붙이는 방식으로 사용하면 기존 객체의 삭제를 올바르게 표현할 수 없습니다.

데이터셋 뷰의 예외

데이터셋 뷰(Dataset view)를 입력으로 사용할 때는 다음 조건을 확인합니다.

• 삭제 열이 있는 뷰는 어떤 행이 사라졌는지 증분으로 결정할 수 없어 항상 전체 재인덱싱합니다.
• 중복 제거 순서 열의 큰 값이 더 오래된 트랜잭션에 있으면 증분 범위 밖의 값이 무시될 수 있습니다. 순서 값이 트랜잭션마다 증가하도록 만들거나, 삭제 열을 두어 전체 재인덱싱 경로를 사용해야 합니다.
• 기반 데이터셋의 최신 뷰에서 이전 트랜잭션을 삭제하는 보존 정책은 예상과 다른 증분 결과를 만들 수 있습니다. 이 경우 전체 재인덱싱으로 상태를 다시 계산합니다.

전체 재인덱싱 조건

한 트랜잭션에서 바뀐 행의 비율이 기본 임계값인 80%를 넘으면 전체 재인덱싱이 변경분 계산보다 저렴하고 빠를 수 있습니다. 스키마 변경으로 새 파이프라인과 인덱스가 필요한 경우에도 전체 재인덱싱합니다. 사용자가 Ontology Manager에서 직접 전체 재인덱싱을 시작할 수도 있습니다.

모니터링과 실패 조사

Ontology Manager에서 Monitor the health of this object type을 선택하면 Data Health의 모니터링 뷰로 이동합니다. Live와 Replacement의 각 작업을 감시하고 Sync Propagation Delay 규칙으로 인덱스 데이터의 신선도가 임계값을 넘을 때 알림을 받을 수 있습니다.

구형 OSv1은 동기화 작업 하나에 상태검사를 정의했지만, OSv2는 여러 빌드 작업을 모니터링 뷰로 관리합니다.

기반 데이터소스에 View 권한이 있으면 객체 유형 Datasources의 Live pipeline에서 실패한 노드를 선택하고 Failed job을 열어 오류를 확인할 수 있습니다.

Builds 앱의 왼쪽 검색 필터에서 객체 유형을 지정하여 관련 빌드 작업 전체를 찾는 방법도 있습니다.