Python 저장소에서 미디어 집합 배치 변환 만들기

Python 변환은 미디어 집합의 항목을 경로나 RID로 읽고, 목록을 데이터프레임으로 다루며, 변환한 미디어를 새 미디어 집합에 저장할 수 있습니다. 이 예제는 각 PDF의 첫 페이지를 PNG로 바꿉니다.

저장소와 라이브러리

프로젝트에서 + New > Code repository를 선택합니다. 이름을 정하고 Language template에서 Python을 고른 뒤 Initialize repository를 누릅니다.

입력 미디어 집합을 프로젝트에서 준비합니다. 왼쪽 Libraries에서 transforms-media를 검색하여 Add and install library를 선택합니다. 설치가 끝나면 Library installed 상태를 확인합니다.

자동 생성된 examples.py를 사용하거나 ... > Rename으로 이름을 바꿀 수 있습니다. 새 파일이 필요하면 Datasets 아래에서 ... > New file을 선택하여 Python 파일을 만듭니다.

미디어 입력과 출력 준비

입력은 transforms.mediasets.MediaSetInput, 출력은 MediaSetOutput으로 선언하고 transform 계열 데코레이터를 사용합니다. 아래 예제는 Spark 실행을 명시하는 @transform.spark.using을 사용합니다. 빌드 시 선언은 각각 미디어 집합에 접근할 수 있는 런타임 입력 및 출력 객체로 바뀝니다.

예제의 출력 미디어 집합은 프로젝트의 New > Media set에서 먼저 만듭니다. PNG를 쓸 대상이라면 이미지에 맞는 미디어 집합 구성을 선택합니다.

Code Repositories에서 아직 없는 출력 경로의 전구 메뉴로 Create media set을 선택하는 방법도 있습니다. 미디어 집합 스키마와 구성을 입력하면 MediaSetOutput에 생성 설정이 들어가고 첫 빌드에서 미디어 집합이 만들어집니다. 생성 후에는 그 초기화 필드를 임의로 수정하지 않습니다. 아래 코드의 경로는 미리 준비한 실제 입력과 출력으로 바꿉니다.

개별 항목 읽기와 쓰기

get_media_item_by_path()는 미디어 집합 안의 경로로, get_media_item()은 항목 RID로 읽습니다. 반환값은 Python의 파일 유사 스트림이며 io.open()에 전달하는 옵션을 지원합니다. 순차 스트림이므로 임의 위치 탐색에 의존하지 않습니다.

다음 예제는 경로로 읽은 항목을 다른 미디어 집합에 복사합니다. 입력에 image1.jpg가 있어야 합니다.

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


@transform.spark.using(
    images=MediaSetInput("/examples/images"),
    output_images=MediaSetOutput("/examples/output_images"),
)
def copy_image(images, output_images):
    with images.get_media_item_by_path("image1.jpg") as source:
        output_images.put_media_item(source, "copied_image1.jpg")

RID를 사용하는 경우에는 실제 런타임 입력과 출력, 항목 RID 및 출력 경로를 받는 다음 함수를 변환에서 호출할 수 있습니다.

def copy_item_by_rid(media_input, media_output, media_item_rid, output_path):
    with media_input.get_media_item(media_item_rid) as source:
        return media_output.put_media_item(source, output_path)

put_media_item()은 파일 유사 객체와 출력 미디어 항목의 경로를 받습니다. 미디어를 변환하지 않는 단순 복사에는 출력의 fast_copy_media_item(media_input, media_item_rid, output_path)를 사용할 수 있습니다. 원천 미디어 입력, 실제 항목 RID와 출력 경로를 전달하며 항목을 내려받아 다시 올리는 과정을 줄입니다.

미디어 항목 목록을 표로 저장

list_media_items_by_path_with_media_reference(ctx)는 항목 목록을 Spark 데이터프레임으로 반환합니다. 같은 경로에 여러 항목이 있으면 최신 항목만 목록에 들어갑니다.

미디어 참조로 읽을 수 있게 typeclass도 함께 기록합니다.

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


@transform.spark.using(
    images=MediaSetInput("/examples/images"),
    listing_output=Output("/examples/listed_images"),
)
def list_images(ctx, images, listing_output):
    listing = images.list_media_items_by_path_with_media_reference(ctx)
    listing_output.write_dataframe(
        listing,
        column_typeclasses={
            "mediaReference": [{"kind": "reference", "name": "media_reference"}]
        },
    )

이 데이터프레임에 일반 PySpark 연산을 적용하여 처리할 항목을 선택할 수 있습니다.

PDF 첫 페이지를 PNG로 저장

다음 코드는 입력 목록을 순회하며 0번 페이지인 첫 페이지를 PNG로 변환합니다.

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


@transform.spark.using(
    pdfs=MediaSetInput("/examples/PDFs"),
    output_pngs=MediaSetOutput("/examples/PNGs"),
)
def upload_first_pages(ctx, pdfs, output_pngs):
    listing = pdfs.list_media_items_by_path_with_media_reference(ctx)
    for item in listing.select("mediaItemRid").toLocalIterator():
        with pdfs.transform_document_to_png(item.mediaItemRid, 0) as png:
            output_pngs.put_media_item(png, f"PNG_{item.mediaItemRid}.png")

이 예제는 드라이버에서 항목을 순차 처리합니다. toLocalIterator()는 모든 RID를 한 목록으로 수집하지 않지만 가장 큰 파티션을 처리할 메모리는 필요합니다. 데이터 규모와 변환 종류에 맞춰 병렬 처리 및 미디어 사용 한도를 고려합니다.

브랜치에서 확인하고 운영에 반영

미디어 집합도 브랜치에서 다룰 수 있으므로 운영 입력과 출력을 바로 바꾸지 않고 변환을 확인할 수 있습니다. 미디어 집합을 사용하는 변환은 Code Repositories의 Preview 기능으로 실행할 수 없으므로 실제 브랜치 빌드에서 결과를 확인합니다.

편집 내용은 기본적으로 자동 저장됩니다. Commit은 변경 묶음을 기록하고 자동 검사를 실행합니다. Build는 커밋 및 검사를 수행하면서 출력을 빌드합니다. 검사만 수행하려면 Commit을 사용합니다.

브랜치의 출력 항목과 형식이 맞는지 확인한 뒤 동료와 변경을 검토하고 병합합니다. 운영 브랜치에서도 빌드하여 새 미디어 출력이 반영되었는지 확인합니다. 일반적인 검토 및 병합 절차는 저장소 배치 파이프라인에서 설명합니다.