미디어 집합의 증분 읽기와 쓰기

미디어 입력과 출력 모두를 증분으로 사용하려면 @incremental(v2_semantics=True)를 지정합니다. Spark 변환에서는 다른 증분 데이터셋이나 가상 테이블 입출력과 함께 사용할 수 있으며, 모든 입력과 출력이 해당 실행의 증분 가능성 판단에 참여합니다.

추가된 항목만 복사하기

다음 예제의 출력은 transactional 이미지 미디어 집합입니다. 증분 실행에서는 새 입력을 읽고 modify로 추가하며, 비증분 실행에서는 전체 입력을 읽고 replace로 교체합니다.

from transforms.api import incremental, transform
from transforms.mediasets import MediaSetInput, MediaSetOutput


@incremental(v2_semantics=True)
@transform.spark.using(
    images=MediaSetInput("/Example Project/Inputs/pngs"),
    copied=MediaSetOutput("/Example Project/Outputs/copied_pngs"),
)
def compute(images, copied):
    def copy_item(row):
        copied.fast_copy_media_item(images, row.media_item_rid, row.path)

    images.dataframe().foreach(copy_item)

증분 조건을 깨뜨리는 변경

미디어 출력이 다음 상태이면 증분 실행을 막을 수 있습니다.

• 다중 출력에서 다른 출력과 달리 마지막 빌드를 다른 변환이 수행한 경우
• Transactional 출력에 마지막 빌드 이후 사용자가 업로드하거나 항목을 삭제하는 등 변경을 가한 경우

미디어 입력의 내용을 replace로 교체하면 일반적으로 증분 조건이 깨집니다. 단, 입력을 snapshot_inputs에 등록하면 전체 교체 자체가 증분 실행을 막지 않습니다.

데이터셋과 달리 미디어 입력의 경로 덮어쓰기나 항목 삭제는 그 사실만으로 증분 실행을 막지 않습니다. 알고리즘이 기존 출력의 수정이나 삭제를 반영해야 하는 경우에는 별도 계산이 필요합니다. 다른 데이터셋 입력이 증분 조건을 충족하지 못하면 미디어 입력이 유효해도 변환 전체는 비증분으로 실행될 수 있습니다.

읽기 모드와 경로 중복

항목 수준의 added와 previous 합집합은 current와 같습니다. 비증분 실행에서는 previous가 빈 목록이며 이전 빌드 당시 항목을 복원하는 모드로 사용할 수 없습니다.

기본값은 증분 실행에서 added, 비증분 실행에서 current입니다. 목록 메서드의 mode로 명시할 수 있습니다. dataframe(deduplicate_by_path=False, mode="previous")는 경로별 중복 제거 없이 previous 범위를 읽습니다.

경로 중복 제거를 켜면 같은 경로의 가장 최근 항목만 나옵니다. 같은 경로에 추가된 모든 버전을 처리하려면 deduplicate_by_path=False를 사용합니다.

경로가 덮어쓰였을 때 전체를 다시 쓰기

다음 구현은 transactional 출력에 사용합니다. 이전 범위와 새 범위에 같은 경로가 있으면 현재 전체 입력을 읽고 출력 전체를 교체합니다. 이는 코드가 읽기/쓰기 전략을 고르는 것이며 실행 문맥의 ctx.is_incremental 값을 변경하는 것은 아닙니다.

from transforms.api import incremental, transform
from transforms.mediasets import MediaSetInput, MediaSetOutput


@incremental(v2_semantics=True)
@transform.spark.using(
    images=MediaSetInput("/Example Project/Inputs/pngs"),
    copied=MediaSetOutput("/Example Project/Outputs/reconciled_pngs"),
)
def compute(ctx, images, copied):
    if not ctx.is_incremental:
        read_mode = "current"
        write_mode = "replace"
    else:
        previous = images.dataframe(deduplicate_by_path=False, mode="previous")
        added = images.dataframe(deduplicate_by_path=False, mode="added")
        overwritten = bool(
            previous.select("path").join(added.select("path"), on="path", how="inner").head(1)
        )
        read_mode = "current" if overwritten else "added"
        write_mode = "replace" if overwritten else "modify"
    copied.set_write_mode(write_mode)

    def copy_item(row):
        copied.fast_copy_media_item(images, row.media_item_rid, row.path)

    images.dataframe(mode=read_mode).foreach(copy_item)

처음 실행하거나 증분 조건이 깨진 실행도 replace로 처리합니다. 여기서는 최종 복사 목록의 기본 경로 중복 제거를 사용하여 각 경로의 현재 항목을 출력합니다. 입력 삭제 자체를 탐지하여 이전 출력에서 지우는 알고리즘은 이 경로 덮어쓰기 검사와 별개입니다.

새 브랜치와 transactionless 출력

미디어 집합은 증분 fallback branch를 지원하지 않습니다. 새 브랜치에서 출력이 비어 있으면 incremental decorator가 스냅샷을 권장할 수 있습니다. 같은 코드를 main에서 빌드한다고 반드시 같은 스냅샷 판단이 나오는 것은 아닙니다.

Transactionless 미디어 집합은 modify만 지원하고 replace로 스냅샷을 만들 수 없습니다. 이런 집합이 증분 출력인데 변환이 증분으로 실행될 수 없으면 빌드가 실패합니다. 다른 입력의 전체 교체, 출력 변경과 브랜치 상태 등 증분 불가 원인을 조사해야 합니다.

미디어 출력 하나만 개별적으로 abort할 수는 없습니다. 작업을 중단하려면 ctx.abort_job()으로 전체 출력을 함께 중단하여 다음 실행의 증분 이력을 일관되게 유지합니다.

항목 수로 배치 제한하기

미디어 입력의 batch_limit은 작업마다 처리할 항목 수의 상한이며 Spark 변환에서만 지원합니다. 데이터셋 입력의 transaction_limit이 트랜잭션 수를 세는 것과 구분합니다.

제한은 경로 중복 제거보다 먼저 적용합니다. 예를 들어 limit 100으로 읽은 뒤 경로 중복을 제거하면 결과 목록은 100개보다 적을 수 있습니다. 스냅샷 재계산이나 오랫동안 쌓인 입력도 여러 작업으로 나누어 처리할 수 있습니다.

다음은 같은 이미지 schema의 입력 세 개를 transactional 출력으로 복사하는 예제입니다. 한 작업에서 첫 입력은 최대 100개, 둘째는 최대 500개, 셋째는 제한 없이 읽습니다. 모든 버전을 처리하고 서로 다른 소스와 항목을 구분하도록 출력 경로에 소스 이름과 RID를 포함합니다.

from transforms.api import incremental, transform
from transforms.mediasets import MediaSetInput, MediaSetOutput


@incremental(v2_semantics=True, strict_append=True)
@transform.spark.using(
    first=MediaSetInput("/Example Project/Inputs/first_images", batch_limit=100),
    second=MediaSetInput("/Example Project/Inputs/second_images", batch_limit=500),
    third=MediaSetInput("/Example Project/Inputs/third_images"),
    output=MediaSetOutput("/Example Project/Outputs/combined_images"),
)
def compute(first, second, third, output):
    for label, media in [("first", first), ("second", second), ("third", third)]:
        def copy_item(row, source=media, prefix=label):
            path = f"{prefix}/{row.media_item_rid}/{row.path}"
            output.fast_copy_media_item(source, row.media_item_rid, path)

        media.dataframe(deduplicate_by_path=False).foreach(copy_item)

한 배치의 성공이 전체 입력을 따라잡았다는 뜻은 아닙니다. 성공 후 재실행 스케줄의 절차로 출력이 최신이 될 때까지 빌드하도록 구성합니다. 새 데이터가 계속 도착하면 재실행도 계속될 수 있으며 Force build와 재실행 옵션을 함께 사용하지 않습니다.