Java 변환 정의, 등록과 실행 문맥

Transform은 입력 및 출력 데이터셋, 실제 계산 함수와 런타임 프로필 같은 추가 구성을 묶습니다. 이를 Pipeline에 등록해야 Foundry가 출력의 빌드 방법을 찾고 실행할 수 있습니다. 별도 증분 설정이 없는 변환의 기본 트랜잭션은 SNAPSHOT입니다.

고수준과 저수준 API

계약	고수준	저수준
표 처리	`Dataset<Row>`로 직접 처리	입력 핸들에서 `asDataFrame().read()`로 읽기
파일 접근	지원하지 않음	지원
입력 수	여러 개	여러 개
출력 수	하나	여러 개
계산 결과	`Dataset<Row>`를 반환	출력 핸들에 직접 쓰며 반환값은 `void`

계산 함수는 클래스 안의 public, 비정적 메서드여야 하고 @Compute가 필요합니다. Annotation이 없으면 계산 함수가 올바르게 등록되지 않습니다.

아래 예제는 각 파일의 이름과 패키지를 맞춰 Java 소스 폴더에 둡니다. /Example Project/... 경로들은 실제 프로젝트 입출력으로 바꿔야 합니다.

자동 등록의 고수준 변환

입력 매개변수마다 @Input, 반환 출력에는 메서드의 @Output을 붙입니다. 다음 파일은 myproject/datasets/HighLevelAutoTransform.java입니다.

package myproject.datasets;

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

public final class HighLevelAutoTransform {
    @Compute
    @Output("/Example Project/Outputs/high_auto")
    public Dataset<Row> compute(
            @Input("/Example Project/Inputs/source") Dataset<Row> input) {
        return input.limit(10);
    }
}

자동 등록의 저수준 변환

입력은 FoundryInput, 출력은 FoundryOutput으로 받습니다. 각 인자에 해당 annotation을 붙입니다. 다음 파일은 myproject/datasets/LowLevelAutoTransform.java입니다.

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 org.apache.spark.sql.Dataset;
import org.apache.spark.sql.Row;

public final class LowLevelAutoTransform {
    @Compute
    public void compute(
            @Input("/Example Project/Inputs/source") FoundryInput input,
            @Output("/Example Project/Outputs/low_auto") FoundryOutput output) {
        Dataset<Row> limited = input.asDataFrame().read().limit(10);
        output.getDataFrameWriter(limited).write();
    }
}

Pipeline.autoBindFromPackage("myproject.datasets")는 필요한 입출력 annotation이 있는 정의만 발견합니다. 같은 패키지에 있어도 필수 annotation이 없으면 자동 등록하지 않습니다.

대부분의 작업에는 자동 등록을 권장합니다. @StopPropagating, @StopRequiring 같은 보안 관련 annotation도 자동 등록에서만 지원합니다.

수동 등록의 계산 클래스

입출력 조합을 메타프로그래밍으로 반복 정의하거나 등록 대상을 명시적으로 제어할 때는 수동 등록을 사용합니다. 계산 클래스에는 @Compute만 두고 자산 바인딩은 PipelineDefiner에서 지정합니다.

고수준 계산 파일 myproject/datasets/HighLevelManualFunction.java:

package myproject.datasets;

import com.palantir.transforms.lang.java.api.Compute;
import org.apache.spark.sql.Dataset;
import org.apache.spark.sql.Row;

public final class HighLevelManualFunction {
    @Compute
    public Dataset<Row> compute(Dataset<Row> myInput) {
        return myInput.limit(10);
    }
}

저수준 계산 파일 myproject/datasets/LowLevelManualFunction.java:

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 org.apache.spark.sql.Dataset;
import org.apache.spark.sql.Row;

public final class LowLevelManualFunction {
    @Compute
    public void compute(FoundryInput myInput, FoundryOutput myOutput) {
        Dataset<Row> limited = myInput.asDataFrame().read().limit(10);
        myOutput.getDataFrameWriter(limited).write();
    }
}

하나의 PipelineDefiner로 등록

Java 하위 프로젝트마다 PipelineDefiner 구현 하나를 둡니다. 다음은 자동 등록 대상과 두 수동 정의를 함께 등록하는 myproject/MyPipelineDefiner.java입니다.

package myproject;

import com.palantir.transforms.lang.java.api.HighLevelTransform;
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.datasets.HighLevelManualFunction;
import myproject.datasets.LowLevelManualFunction;

public final class MyPipelineDefiner implements PipelineDefiner {
    @Override
    public void define(Pipeline pipeline) {
        pipeline.autoBindFromPackage("myproject.datasets");

        HighLevelTransform highLevel = HighLevelTransform.builder()
                .computeFunctionInstance(new HighLevelManualFunction())
                .putParameterToInputAlias(
                        "myInput", "/Example Project/Inputs/source")
                .returnedAlias("/Example Project/Outputs/high_manual")
                .build();
        pipeline.register(highLevel);

        LowLevelTransform lowLevel = LowLevelTransform.builder()
                .computeFunctionInstance(new LowLevelManualFunction())
                .putParameterToInputAlias(
                        "myInput", "/Example Project/Inputs/source")
                .putParameterToOutputAlias(
                        "myOutput", "/Example Project/Outputs/low_manual")
                .build();
        pipeline.register(lowLevel);
    }
}

putParameterToInputAlias와 putParameterToOutputAlias의 첫 문자열은 계산 함수의 매개변수 이름과 정확히 같아야 합니다. 고수준의 반환 출력은 returnedAlias로 지정합니다. 저수준은 반환값이 없으므로 출력 인자마다 매핑합니다. register() 호출 하나는 변환 하나를 등록합니다.

진입점은 Java ServiceLoader로 찾습니다. src/main/resources/META-INF/services/com.palantir.transforms.lang.java.api.PipelineDefiner에 구현의 전체 이름 하나를 기록합니다.

myproject.MyPipelineDefiner

이 파일은 Java 코드가 아닌 서비스 설정 파일입니다. 주석이 필요하면 Java의 //가 아니라 ServiceLoader의 # 문법을 사용합니다. Foundry의 Java 하위 프로젝트는 단일 PipelineDefiner를 요구합니다.

계산 함수에 TransformContext 인자를 추가하면 현재 Spark session, job spec의 transform parameters, 인증 header와 ServiceDiscovery를 받을 수 있습니다. ServiceDiscovery는 발견한 Foundry 서비스의 URI를 제공합니다. 인증 정보는 실행 권한을 가진 자격 증명이므로 로그나 출력 데이터에 기록하지 않습니다.

고수준에서 정수 매개변수 사용

다음 myproject/datasets/ContextHighLevel.java는 job spec에 정수 limit이 구성되어 있어야 합니다. 입력과 출력 경로를 실제 자산에 맞춥니다.

package myproject.datasets;

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

public final class ContextHighLevel {
    @Compute
    @Output("/Example Project/Outputs/context_high")
    public Dataset<Row> compute(
            @Input("/Example Project/Inputs/source") Dataset<Row> input,
            TransformContext context) {
        int limit = (int) context.parameters().get("limit");
        return input.limit(limit);
    }
}

저수준에서 같은 문맥 사용

파일 myproject/datasets/ContextLowLevel.java도 같은 정수 매개변수를 요구합니다.

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.TransformContext;

public final class ContextLowLevel {
    @Compute
    public void compute(
            @Input("/Example Project/Inputs/source") FoundryInput input,
            @Output("/Example Project/Outputs/context_low") FoundryOutput output,
            TransformContext context) {
        int limit = (int) context.parameters().get("limit");
        output.getDataFrameWriter(input.asDataFrame().read().limit(limit)).write();
    }
}

두 클래스도 같은 자동 등록 패키지에 두면 발견됩니다. limit이 없는 환경의 임의 기본값을 만드는 대신 실제 job parameter 계약을 준비해야 합니다.

Keyboard shortcuts

팔란티어 플랫폼 한국어 해설