미디어 업로드, 빠른 복사와 쓰기 모드

MediaSetOutput의 실제 출력 핸들에 파일 유사 객체와 출력 경로를 전달하면 미디어 항목을 업로드할 수 있습니다. 출력 schema는 생성할 미디어 형식과 맞아야 합니다.

스트림 업로드와 빠른 복사

put_media_item(stream, path)는 스트림을 읽어 새 항목을 업로드합니다. 같은 미디어를 그대로 다른 집합에 복사할 때는 다운로드와 재업로드를 생략하는 fast_copy_media_item()이 더 효율적입니다.

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


@transform.spark.using(
    images=MediaSetInput("/Example Project/Inputs/images"),
    copied=MediaSetOutput("/Example Project/Outputs/copied_images"),
)
def compute(images, copied):
    rid = images.get_media_item_rid_by_path("image1.jpg").item
    copied.fast_copy_media_item(images, rid, "fast_copied_image1.jpg")

경로로 RID를 조회한 결과와 미디어의 바이트 스트림은 구분합니다. 실제 내용을 읽어 처리해야 할 때는 get_media_item_by_path() 또는 get_media_item()으로 스트림을 엽니다.

PDF의 첫 페이지를 JPEG로 업로드하고 참조 저장하기

다음 변환은 PDF마다 0번 페이지를 JPEG로 만들고 이미지 미디어 집합에 업로드합니다. 입력은 PDF 집합이고 출력 미디어 집합은 JPEG를 받아야 합니다. 전체 문서의 모든 페이지를 변환하는 예제는 아닙니다.

from pyspark.sql import functions as F
from pyspark.sql.types import StringType
from transforms.api import Output, transform
from transforms.mediasets import MediaSetInput, MediaSetOutput

REFERENCE_TYPES = {
    "mediaReference": [{"kind": "reference", "name": "media_reference"}]
}


@transform.spark.using(
    pdfs=MediaSetInput("/Example Project/Inputs/pdfs"),
    images=MediaSetOutput("/Example Project/Outputs/first_page_jpegs"),
    references=Output("/Example Project/Outputs/jpeg_references"),
)
def compute(ctx, pdfs, images, references):
    def upload_first_page(rid, path):
        with pdfs.transform_document_to_jpg(rid, 0) as jpeg:
            response = images.put_media_item(jpeg, f"{path}.page-0.jpg")
        return response.media_item_rid

    upload = F.udf(upload_first_page, StringType()).asNondeterministic()
    listing = pdfs.list_media_items_by_path(ctx)
    template = images.media_reference_template()
    result = (
        listing.withColumn(
            "uploaded_media_item_rid", upload(F.col("mediaItemRid"), F.col("path"))
        )
        .select("path", "uploaded_media_item_rid")
        .withColumn(
            "mediaReference", F.format_string(template, F.col("uploaded_media_item_rid"))
        )
    )
    references.write_dataframe(result, column_typeclasses=REFERENCE_TYPES)

출력 이름은 원래 경로에 .page-0.jpg를 덧붙여 원본 경로를 구분합니다. UDF가 실제 업로드를 수행하므로 Spark 작업 재시도나 별도 action으로 다시 평가되는 상황에서 호출이 반복될 수 있음을 고려합니다. 미디어 참조 문자열은 출력 핸들의 template과 실제 업로드 응답 RID로 만듭니다.

modify와 replace

Transactional 미디어 집합의 기본값은 replace입니다. Transactionless 집합은 modify만 사용하며 브랜치를 비우는 replace로 바꿀 수 없습니다.

set_write_mode()는 첫 항목을 업로드하기 전까지 호출할 수 있습니다. 다음 함수는 실제 Catalog 데이터셋 입력과 미디어 출력 핸들, 호출자가 정한 replace_output 불리언을 받습니다. 출력이 transactional인 경우에 두 정책을 모두 선택할 수 있습니다.

def upload_with_policy(input_dataset, output_media_set, *, replace_output: bool):
    output_media_set.set_write_mode("replace" if replace_output else "modify")
    output_media_set.put_dataset_files(input_dataset)

정책 결정과 실제 업로드를 분리하며, 업로드한 뒤에는 같은 실행에서 쓰기 모드를 변경하지 않습니다.

Catalog 데이터셋의 파일 일괄 업로드

put_dataset_files()는 Spark 분산 처리를 사용하여 일반 파일 데이터셋의 파일을 미디어 집합에 업로드합니다. Lightweight에서는 지원하지 않습니다.

from transforms.api import Input, transform
from transforms.mediasets import MediaSetOutput


@transform.spark.using(
    pdf_files=Input("/Example Project/Inputs/raw_pdfs"),
    pdf_media=MediaSetOutput("/Example Project/Outputs/pdf_media"),
)
def compute(pdf_files, pdf_media):
    pdf_media.put_dataset_files(pdf_files, ignore_items_not_matching_schema=False)

ignore_items_not_matching_schema=False이면 PDF 집합에 JPEG가 섞인 경우처럼 schema와 맞지 않는 항목이 있을 때 빌드가 실패합니다. True로 지정하면 불일치 항목을 건너뜁니다. 이 선택은 모든 파일을 반드시 처리해야 하는지에 따라 명시적으로 정합니다.

파일별 업로드

파일마다 별도 경로를 정해야 하면 put_media_item()을 사용합니다. 다음 예제는 기존 항목을 유지하는 should_snapshot=False 선언을 사용하고 각 파일의 basename을 출력 경로로 정합니다. 입력의 서로 다른 디렉터리에 같은 basename이 없어야 이 경로 규칙으로 항목을 구분할 수 있습니다.

import os

from transforms.api import Input, transform
from transforms.mediasets import MediaSetOutput


@transform.spark.using(
    files=Input("/Example Project/Inputs/raw_media_files"),
    media=MediaSetOutput("/Example Project/Outputs/media", should_snapshot=False),
)
def compute(files, media):
    filesystem = files.filesystem()
    for status in filesystem.ls():
        with filesystem.open(status.path, "rb") as stream:
            media.put_media_item(stream, os.path.basename(status.path))

디렉터리 구분도 보존하는 경로 정책을 사용한다면 status.path를 업로드 경로로 전달할 수 있습니다. 증분 처리를 적용할 때는 미디어 집합의 트랜잭션 정책과 기본 읽기/쓰기 모드를 함께 확인합니다.