Spark 프로필 가져오기와 코드에 적용하기

Spark 프로필을 변환에 사용하려면 먼저 해당 프로젝트에 프로필 참조를 가져와야 합니다. 등록하지 않은 프로필을 코드에서 사용하면 변환을 게시하는 Checks가 실패합니다.

프로젝트에 프로필 가져오기

Code Repositories의 Settings > Spark에서 Enabled profiles를 확인합니다. Add profiles를 선택하고 필요한 프로필을 검색합니다.

기본적으로 Resource Management 관리자가 제한 프로필을 가져올 수 있습니다. 환경에 spark-profile-admins 그룹이 구성되어 있을 수도 있습니다. 일반 사용자는 필요한 프로필을 관리자에게 요청합니다.

프로필 위에 포인터를 놓고 Import를 선택하면 프로젝트 참조로 추가됩니다. 가져온 프로필은 그 프로젝트의 모든 저장소에서 사용할 수 있으며, 프로젝트 Summary 사이드바의 References에서도 확인할 수 있습니다.

적용 순서

프로필은 왼쪽부터 오른쪽으로 평가합니다. 같은 Spark 설정을 여러 프로필이 지정하면 목록의 뒤쪽 값이 우선합니다. 이는 개별 설정 키의 우선순위이므로 프로필이 설정하는 모든 키를 전체 참조에서 확인합니다.

아래 예제에서는 프로젝트에 EXECUTOR_MEMORY_MEDIUM과 NUM_EXECUTORS_4를 가져왔다고 가정합니다. 데이터셋 경로는 실제 입력과 출력으로 지정합니다.

Python

Spark 변환의 with_config(profiles=...)에 프로필 목록을 전달합니다.

from transforms.api import Input, Output, transform


@transform.spark.using(
    my_input=Input("/examples/profiles/input"),
    my_output=Output("/examples/profiles/python_output"),
).with_config(profiles=["EXECUTOR_MEMORY_MEDIUM", "NUM_EXECUTORS_4"])
def copy_with_profiles(my_input, my_output):
    my_output.write_dataframe(my_input.dataframe())

Java 자동 등록

자동 등록은 계산 메서드의 @TransformProfiles에 프로필 배열을 지정합니다. 해당 패키지가 기존 PipelineDefiner의 자동 등록 대상이어야 합니다.

package myproject.datasets;

import com.palantir.transforms.lang.java.api.Compute;
import com.palantir.transforms.lang.java.api.FoundryInput;
import com.palantir.transforms.lang.java.api.FoundryOutput;
import com.palantir.transforms.lang.java.api.Input;
import com.palantir.transforms.lang.java.api.Output;
import com.palantir.transforms.lang.java.api.TransformProfiles;
import org.apache.spark.sql.Dataset;
import org.apache.spark.sql.Row;

public final class ProfiledAutoTransform {
    @Compute
    @TransformProfiles({"EXECUTOR_MEMORY_MEDIUM", "NUM_EXECUTORS_4"})
    public void compute(
            @Input("/examples/profiles/input") FoundryInput input,
            @Output("/examples/profiles/java_auto_output") FoundryOutput output) {
        Dataset<Row> limited = input.asDataFrame().read().limit(10);
        output.getDataFrameWriter(limited).write();
    }
}

Java 수동 등록

수동 등록에서는 변환 빌더의 transformProfiles()에 목록을 전달합니다. 계산 함수를 다음처럼 별도 클래스로 정의할 수 있습니다.

package myproject.manual;

import com.palantir.transforms.lang.java.api.Compute;
import com.palantir.transforms.lang.java.api.FoundryInput;
import com.palantir.transforms.lang.java.api.FoundryOutput;

public final class ProfiledManualFunction {
    @Compute
    public void compute(FoundryInput myInput, FoundryOutput myOutput) {
        myOutput.getDataFrameWriter(myInput.asDataFrame().read().limit(10)).write();
    }
}

기존 Java 하위 프로젝트의 PipelineDefiner에서 이 함수를 등록합니다. 예제 전체는 다음과 같습니다.

package myproject;

import com.google.common.collect.ImmutableList;
import com.palantir.transforms.lang.java.api.LowLevelTransform;
import com.palantir.transforms.lang.java.api.Pipeline;
import com.palantir.transforms.lang.java.api.PipelineDefiner;
import myproject.manual.ProfiledManualFunction;

public final class MyPipelineDefiner implements PipelineDefiner {
    @Override
    public void define(Pipeline pipeline) {
        LowLevelTransform transform = LowLevelTransform.builder()
            .transformProfiles(ImmutableList.of(
                "EXECUTOR_MEMORY_MEDIUM", "NUM_EXECUTORS_4"))
            .computeFunctionInstance(new ProfiledManualFunction())
            .putParameterToInputAlias("myInput", "/examples/profiles/input")
            .putParameterToOutputAlias("myOutput", "/examples/profiles/java_manual_output")
            .build();
        pipeline.register(transform);
    }
}

서비스 로더 파일 resources/META-INF/services/com.palantir.transforms.lang.java.api.PipelineDefiner에는 사용할 구현의 정규 클래스 이름을 한 개 기록합니다. 위 예제라면 myproject.MyPipelineDefiner입니다. 기존 구현이 있으면 그 정의에 등록을 추가하며 여러 구현을 동시에 만들지 않습니다.

SQL

테이블의 foundry_transform_profiles 속성에 쉼표로 구분한 프로필 이름을 지정합니다.

CREATE TABLE `/examples/profiles/sql_properties_output`
TBLPROPERTIES (
    foundry_transform_profiles = 'EXECUTOR_MEMORY_MEDIUM, NUM_EXECUTORS_4'
)
AS SELECT * FROM `/examples/profiles/input`

형식과 옵션을 지정하는 다음 문법도 사용할 수 있습니다.

CREATE TABLE `/examples/profiles/sql_options_output`
USING parquet
OPTIONS (
    foundry_transform_profiles = 'EXECUTOR_MEMORY_MEDIUM, NUM_EXECUTORS_4'
)
AS SELECT * FROM `/examples/profiles/input`

사용자 지정 변환 프로필 지정은 Foundry의 ANSI SQL 모드에서는 지원하지 않습니다.

적용 후 판단

프로필은 필요한 자원만 조정하는 것이 좋습니다. Spark 자원 기준에 따라 한 번에 한 항목을 바꾸고, Spark details에서 효과를 확인합니다. 과거 실행의 실제 사용량에 따라 자동으로 자원을 조정하려면 관리형 프로필을 사용할 수 있습니다.