トランスフォームとパイプライン

Transform は、データセットの計算方法を記述したものです。次の内容を記述します。

• 入力データセットと出力データセット
• 入力データセットを出力データセットに加工するためのコード（計算関数と呼びます）
• 追加の設定（実行時に使用するカスタム Transforms プロファイルなど）

現在、すべてのトランスフォームのトランザクションタイプのデフォルトは SNAPSHOT です。

入力データセットと出力データセット、およびデータ加工コードは、Transform オブジェクトで指定し、Pipeline に登録します。Transform の定義方法は、次の2つの要因によって異なります。

• 高レベルと低レベルのどちらのトランスフォームを定義するか
• プロジェクトのパイプラインにトランスフォームを追加するために、自動登録と手動登録のどちらを使用するか

トランスフォームの種類

DataFrame オブジェクトを使用するデータ加工では、次のいずれかの方法を使用できます。

• Dataset<Row> 型の入力と出力をサポートする高レベルのトランスフォームを定義する
• 低レベルのトランスフォームを定義し、メソッドを明示的に呼び出して、入力データセットを含む Dataset<Row> にアクセスする

ファイルを使用するデータ加工では、低レベルのトランスフォームを定義し、データセット内のファイルにアクセスする必要があります。

2種類のトランスフォームの主な違いを以下にまとめます。

* DataFrame オブジェクトに依存するデータ加工には、高レベルのトランスフォームを使用することを推奨します。

どちらの Transform の種類でも、計算関数を含むクラスを作成する必要があります。このクラス内の計算関数は、@Compute 注釈が付いた public の非 static メソッドである必要があります。この注釈がないと、データ加工コードは正しく登録されません。

登録の種類

リポジトリ内の各 Transforms Java サブプロジェクトは、1個の Pipeline オブジェクトを公開します。この Pipeline オブジェクトは、次の用途に使用します。

1. ビルド方法の指示とともにデータセットを Foundry に登録する
2. Foundry のビルド中に、対象のデータセットをビルドする Transform オブジェクトを見つけて実行する

エントリーポイント

Java のデータ加工を実行するランタイムは、プロジェクトの Pipeline を見つけられる必要があります。Transforms Java は、サービス読み込みのための Java の標準機能 ↗を使用します。

プロジェクトに関連付けられた Pipeline オブジェクトを定義するには、PipelineDefiner オブジェクトを実装する必要があります。この PipelineDefiner オブジェクト内で、プロジェクトのパイプラインにトランスフォームを追加できます。具体的には、各 Java サブプロジェクトで1個の PipelineDefiner オブジェクトを実装する必要があります。

Copied!
1
2
3
4
5
6
7
8
9
10
11
12
13
package myproject;

import com.palantir.transforms.lang.java.api.Pipeline;
import com.palantir.transforms.lang.java.api.PipelineDefiner;

public final class MyPipelineDefiner implements PipelineDefiner {

  @Override
  public void define(Pipeline pipeline) {
    // プロジェクトの Pipeline にトランスフォームを追加するコードをここに記述します。
    // 自動登録または手動登録を使用します。
  }
}

Java パッケージを作成し、PipelineDefiner オブジェクトを実装したら、resources/META-INF/services/com.palantir.transforms.lang.java.api.PipelineDefiner を更新して、PipelineDefiner の実装を指すようにする必要があります。

Copied!
1
2
3
4
// これを「PipelineDefiner」の実装クラス名に置き換えます。
// 各 Java サブプロジェクトは「PipelineDefiner」を1つ実装するため、このファイルには
// エントリーを1件だけ含めることができます。
myproject.MyPipelineDefiner

MyPipelineDefiner は、PipelineDefiner の実装のクラス名を指します。

パイプラインにトランスフォームを追加する

プロジェクトのパイプラインに関連付けられた Transform がデータセットを出力として宣言すると、そのデータセットを Foundry でビルドできるようになります。Transform オブジェクトを Pipeline に追加する推奨方法は、手動登録と自動登録の2つです。

自動登録

プロジェクトが複雑になるにつれて、Transform オブジェクトを Pipeline に手動で追加する作業は煩雑になりがちです。そのため、Pipeline オブジェクトには、Java パッケージ内のすべての Transform オブジェクトを見つけるための autoBindFromPackage() メソッドが用意されています。自動登録を使用するには、次の操作が必要です。

• Transform に対応するクラスを定義します。自動登録では、入力データセットと出力データセットの情報、および計算関数を含むクラスを定義します。
• 必要な @Input 注釈と @Output 注釈を追加します。
• Pipeline.autoBindFromPackage() メソッドを呼び出して、指定した Java パッケージ内の Transform 定義を登録します。autoBindFromPackage() メソッドは、必要な注釈が付いているトランスフォーム定義のみを登録します。必要な注釈が付いていないトランスフォームは、autoBindFromPackage() メソッドに指定した Java パッケージ内にあっても、プロジェクトの Pipeline に追加されません。

手動登録

Transform オブジェクトは、Pipeline.register() メソッドを使用して Pipeline に手動で追加できます。このメソッドは、1回の呼び出しで1個の Transform を登録できます。トランスフォームで手動登録を使用するには、次の操作が必要です。

• Transform オブジェクトの計算関数を含むクラスを定義します。自動登録とは異なり、手動登録では、PipelineDefiner の実装内で入力データセットと出力データセットの情報を指定します
• HighLevelTransform.builder() または LowLevelTransform.builder() を使用して、使用する計算関数を指定し、入力データセットと出力データセットを指定します。
• Pipeline.register() メソッドを呼び出して、Transform 定義をプロジェクトのパイプラインに明示的に追加します。

トランスフォームのコンテキスト

データ加工が入力データセット以外のものに依存する場合があります。たとえば、データ加工で現在の Spark セッションや Job Spec 内のトランスフォームパラメーターにアクセスする必要がある場合です。このような場合は、データ加工に TransformContext オブジェクトを注入できます。そのためには、計算関数が TransformContext 型のパラメーターを受け取る必要があります。TransformContext には、Transforms の authHeader、Spark セッション、トランスフォームパラメーター、および ServiceDiscovery オブジェクトが含まれます。ServiceDiscovery クラスは、見つかった Foundry サービスのサービス URI を公開します。

Copied!
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
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;

/**
* これは TransformContext にアクセスする高レベルのトランスフォームの例です。
*/
@Compute
public Dataset<Row> myComputeFunction(Dataset<Row> myInput, TransformContext context) {
    int limit = (int) context.parameters().get("limit");
    return myInput.limit(limit);
}

Copied!
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
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;

/**
* これは TransformContext にアクセスする低レベルのトランスフォームの例です。
*/
@Compute
public void compute(FoundryInput input, FoundryOutput output, TransformContext context) {
    int limit = (int) context.parameters().get("limit");
    output.getDataFrameWriter(input.asDataFrame().read().limit(limit)).write();
}