Python 변환에서 ModelInput 사용

palantir_models.transforms.ModelInput은 게시한 모델을 Python 변환의 입력으로 로드합니다. 어댑터의 transform()은 변환 입력을 모델 API가 선언한 표 형식으로 변환하고 결과를 출력 이름별로 제공합니다. Jupyter 작업 공간의 ModelInput은 작업 공간에서 모델을 가져오는 별도 사용 경로입니다.

생성자

모델 경로 또는 RID를 alias에 지정하고 버전, 브랜치와 실행 환경은 키워드 인자로 설정합니다.

sidecar_resources의 cpus는 float CPU 수, memory_gb는 float GB 메모리, gpus는 int GPU 수입니다.

예를 들어 ModelInput("/Housing/models/price_regressor", branch="testing")은 testing 브랜치의 최신 버전을 선택합니다. 경로와 브랜치는 실제 값으로 바꿉니다. 평가 변환은 이렇게 선택하여 로드한 단일 모델 버전에 결과를 연결합니다.

컨테이너 기반 모델을 변환에서 사용하는 경로에는 모델 버전 RID를 명시해야 합니다. 해당 모델 어댑터 라이브러리도 변환 저장소의 의존성에 추가합니다.

경량 변환에서 추론

Spark 분산 추론이 필요하지 않으면 기본 경량 실행인 @transform.using과 sidecar를 권장합니다. 이 조합에는 palantir_models 0.2010.0 이상이 필요합니다. 다음 예제는 모델 API가 Pandas 입력 하나와 df_out이라는 Pandas 출력 하나를 선언한 경우입니다. 경로는 실제 리소스로 바꿉니다.

from transforms.api import Input, Output, transform, LightweightInput, LightweightOutput
from palantir_models import ModelAdapter
from palantir_models.transforms import ModelInput


# Using use_sidecar=True with @transform.using requires palantir_models version 0.2010.0 or higher.
@transform.using(
    data_in=Input("path/to/input"),
    model_input=ModelInput(
        "path/to/my/model",
        use_sidecar=True # runs the model as a sidecar container
    ),
    out=Output('path/to/output'),
)
def my_transform(data_in: LightweightInput, model_input: ModelAdapter, out: LightweightOutput) -> None:
    # Assuming the model's API has a single Pandas input
    # and a single pandas output named `df_out`.
    inference_results = model_input.transform(data_in)
    predictions = inference_results.df_out
    # Alternatively, you can use the predict method on
    # a Pandas DataFrame instance directly:
    # predictions = model_input.predict(data_in.pandas())
    out.write_pandas(predictions)

transform()은 TransformInput 또는 LightweightInput을 선언된 Pandas/Spark 표로 변환합니다. 예제에서 반환한 결과의 df_out을 출력 데이터셋에 기록합니다. 이미 Pandas DataFrame을 준비했다면 주석의 predict() 직접 호출 경로도 사용할 수 있습니다.

sidecar를 사용하지 않을 때

현재 환경이 어댑터 코드를 로드할 수 있어야 합니다. 다른 저장소에서 생성한 모델은 모델과 함께 패키징된 어댑터 Python 라이브러리를 소비 저장소의 의존성에 추가합니다.

정기 재학습하지 않는 모델은 model_version을 지정하여 의도하지 않은 최신 버전이 운영에 반영되는 일을 줄일 수 있습니다. 버전을 생략하면 실행마다 최신 모델 파일을 가져오지만 다른 저장소의 어댑터 라이브러리 버전을 자동 갱신하지는 않습니다. 모델 페이지의 Inference configuration에서 대응 버전을 확인하고 Libraries에서 선택한 뒤 저장소 검사를 통과시킵니다.

모델을 생성한 저장소 안에서 소비하거나 sidecar를 사용하면 이런 수동 라이브러리 갱신 절차를 피할 수 있습니다.

sidecar의 환경과 자원

sidecar는 모델 생성에 사용한 라이브러리 버전으로 추론하고 선택한 모델 버전에 맞는 어댑터를 자동 로드합니다. 소비 저장소에 라이브러리를 직접 가져오면 두 환경의 제약을 합쳐 새로 해결하므로 다른 버전이 선택될 수 있습니다. use_sidecar=True에서는 predict() 요청을 별도 코드 변경 없이 컨테이너로 전달합니다.

다음은 sidecar에 CPU 2개, 메모리 4 GB와 GPU 1개를 요청하는 경량 추론입니다. Spark에서 드라이버 및 실행기 옆에 sidecar를 사용하는 경우 이 자원은 각 컨테이너에 적용됩니다. GPU에는 프로젝트 큐와 패키지 조건도 필요합니다.

from transforms.api import Input, Output, transform, LightweightInput, LightweightOutput
from palantir_models import ModelAdapter
from palantir_models.transforms import ModelInput


# Using use_sidecar=True with @transform.using requires palantir_models version 0.2010.0 or later.
@transform.using(
    data_in=Input("path/to/input"),
    model_input=ModelInput(
        "path/to/my/model",
        use_sidecar=True,
        sidecar_resources={
            "cpus": 2.0,
            "memory_gb": 4.0,
            "gpus": 1
        }
    ),
    out=Output('path/to/output'),
)
def my_transform(data_in: LightweightInput, model_input: ModelAdapter, out: LightweightOutput) -> None:
    inference_results = model_input.transform(data_in)
    out.write_pandas(inference_results.df_out)

Spark 실행기로 분산 추론

DistributedInferenceWrapper는 Spark UDF로 각 실행기에 모델 추론을 분산합니다. 무거운 모델이나 큰 데이터셋의 배치 추론에 사용할 수 있습니다. 다음 RID는 예시이며 실제 입력 데이터셋, 모델과 출력 데이터셋 RID로 바꿉니다.

from transforms.api import transform, Input, Output, configure
from palantir_models.transforms import ModelInput, DistributedInferenceWrapper

@transform.spark.using(
    input_df=Input("ri.foundry.main.dataset.3cd098b3-aae0-455a-9383-4eec810e0ac0"),
    model_input=ModelInput("ri.models.main.model.5b758039-370c-4cfc-835e-5bd3f213454c"),
    output=Output("ri.foundry.main.dataset.c0a3edbc-c917-4f20-88f1-d797ebf27cb2"),
)
def compute(ctx, input_df, model_input, output):
    model_input = DistributedInferenceWrapper(model_input, ctx, 'auto')
    # Use .predict with the DistributedInferenceWrapper and pass it a Spark DataFrame
    # It handles passing the data as pandas to the model and returns a Spark DataFrame
    predictions = model_input.predict(input_df.dataframe())
    # Write the output as a Spark DataFrame
    output.write_dataframe(predictions)

입력과 반환값은 Spark DataFrame입니다. 래퍼 내부에서는 모델 API의 Pandas 입력과 출력을 사용합니다. 모델 API는 Pandas DataFrame 입력 하나와 출력 하나를 가져야 하며 추가 입력 매개변수는 여러 개 사용할 수 있습니다. sidecar는 선택 사항입니다.

DistributedInferenceWrapper(model, ctx, num_partitions="auto", max_rows_per_chunk=1_000_000)

실행기 수는 Spark 프로필로 구성합니다. 여러 입력 행이 함께 있어야 예측할 수 있는 모델에는 이 래퍼를 사용하지 않습니다. 예를 들어 시계열 이력이 필요한 모델은 같은 시계열의 행이 서로 다른 파티션으로 나뉠 수 있습니다. 필요한 모든 행을 같은 묶음에 보낸다는 보장이 없어 잘못된 예측이 생길 수 있습니다.