注: 以下の翻訳の正確性は検証されていません。AIPを利用して英語版の原文から機械的に翻訳されたものです。

インクリメンタル（差分処理）トランスフォーム

インクリメンタル計算は高度な機能です。この機能を使用する前に、ユーザーガイドのほかの内容を理解していることを確認してください。

ユーザーガイドでこれまでに紹介したトランスフォームは、実行するたびに出力データセット全体を再計算します。これにより、多くの不要な処理が発生する可能性があります。次の例を考えてみます。

Copied!1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
package myproject.datasets;

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

public final class FilterTransform {

   @Compute
   public void myComputeFunction(
           @Input("/examples/students_hair_eye_color") FoundryInput myInput,
           @Output("/examples/students_hair_eye_color_filtered") FoundryOutput myOutput) {
       Dataset<Row> inputDf = myInput.asDataFrame().read();
       myOutput.getDataFrameWriter(inputDf.filter("eye = 'Brown'")).write();
   }
}

/examples/students_hair_eye_color 入力データセットに新しいデータが追加されると、filter() は入力に追加された新しいデータだけでなく、入力全体に対して実行されます。これは計算リソースと時間の浪費になります。

トランスフォームが自身のビルド履歴を把握できれば、出力を計算する方法をより効率的にできます。具体的には、入力に加えられた変更を使用して出力データセットを変更できます。テーブルを再実体化する際にすでに実体化されたデータを使用するこのプロセスを、 インクリメンタル計算と呼びます。インクリメンタル計算を使用しない場合、出力データセットは常にトランスフォームの最新の出力に置き換えられます。

上記のトランスフォームの例に戻ります。このトランスフォームは students データセットに対して filter() を実行し、髪が茶色の学生を書き出します。インクリメンタル計算を使用すると、2人の新しい学生のデータが students に追加された場合、トランスフォームは自身のビルド履歴の情報を使用して、新しい学生のうち髪が茶色の学生だけを出力に追加できます。

RAW                              DERIVED
+---+-----+-----+                  +---+-----+-----+
| id| hair|  eye|                  | id| hair|  eye|
+---+-----+-----+     Build 1      +---+-----+-----+
| 17|Black|Brown|    --------->    | 18|Brown|Brown|
| 18|Brown|Brown|                  +---+-----+-----+
| 19|  Red|Brown|
+---+-----+-----+                            ...
   ...
+---+-----+-----+     Build 2      +---+-----+-----+
| 20|Brown|Brown|    --------->    | 20|Brown|Brown|
| 21|Black|Blue |                  +---+-----+-----+
+---+-----+-----+

インクリメンタルトランスフォームを記述する

transforms-java を使用してインクリメンタルトランスフォームを記述する方法を、順を追って説明します。transforms-python とは異なり、 transforms-java では、注釈を使用してインクリメンタル性を自動的に検証し、変換をインクリメンタルに適用することはありません。 Java でインクリメンタルトランスフォームを記述するプロセスは、ユーザーがより直接的に制御します。ユーザーは、どのような場合に変換をインクリメンタルに実行し、どのような場合には実行しないかを明示的に決定できます。入力データセットがどのように変更されたかを解釈することで、ユーザーは出力データセットをインクリメンタルに更新するか、スナップショットのような方法で更新するかを決定できます。

入力を解釈する

最初のステップは、入力を解釈することです。入力データセットはさまざまな方法で変更される可能性があり、インクリメンタルな変換を適用できるのは特定の状況に限られます。 DataFrameModificationType（または FilesModificationType）は、データセットを変更できるさまざまな方法を表します。モードは次のとおりです。

APPENDED
UPDATED
NEW_VIEW
UNCHANGED

入力がどのように変更されたかに基づいて、入力データセットから何を読み取り、出力データセットに何を書き込むかを決定できます。

入力を読み取る

入力がどのように変更されたかがわかれば、それに応じて読み取れます。トランザクションがデータの追加のみを行った場合は、安全にインクリメンタルに処理し、変更された部分だけを読み取れることが確実です。一方、すでに存在する行の変更を含む入力データセットの変更があった場合は、ビュー全体を再度読み取ることが適切な場合があります。Transforms-Java API では、 readForRange() メソッドによって、入力データセットに対して異なる読み取りモードを使用できます。 ReadRange は、使用できる読み取り範囲を示します。モードは次のとおりです。

UNPROCESSED
PROCESSED
ENTIRE_VIEW

入力の変更タイプを解釈することで、以下の例のように、データをどのように読み取るかを決定できます。

Copied!1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
private ReadRange getReadRange(FoundryInput input) {
     switch (input.asDataFrame().modificationType()) {
         case UNCHANGED:
             LOG.info("No changes in input dataset, read only unprocessed");
             return ReadRange.UNPROCESSED;
         case APPENDED:
             LOG.info("Append-only changes in input dataset, read only unprocessed");
             return ReadRange.UNPROCESSED;
         case UPDATED:
             LOG.info("Update-type changes in input dataset, read entire view");
             return ReadRange.ENTIRE_VIEW;
         case NEW_VIEW:
             LOG.info("New view in input dataset, read entire view");
             return ReadRange.ENTIRE_VIEW;
         default:
             throw new IllegalArgumentException("Unknown ModificationType for input dataset "
                 + input.asDataFrame().modificationType());
     }
}

その後、それに応じて compute メソッドを変更できます。

Copied!1
2
3
4
5
6
7
@Compute
public void myComputeFunction(
        @Input("/examples/students_hair_eye_color") FoundryInput myInput,
        @Output("/examples/students_hair_eye_color_filtered") FoundryOutput myOutput) {
    Dataset<Row> inputDf = myInput.asDataFrame().readForRange(getReadRange(myInput));
    myOutput.getDataFrameWriter(inputDf.filter("eye = 'Brown'")).write();
}

ここまでのコードでは、依然として SNAPSHOT トランザクションが生成されます

この時点では、入力データセットの異なる部分を読み取っているだけで、出力データセットに対する処理は変更していません。この例のここまでのコードを実行すると、入力のどの部分を読み取っているかにかかわらず、出力では常に SNAPSHOT トランザクションが発生します。出力データセットを正しく変更する方法を理解するため、インクリメンタルトランスフォームを適用する前に、チュートリアルを最後まで進めてください。

データを加工する

このステップでは、ユーザーが必要なデータの変換を適用します。入力の変更に応じて、読み取られるデータが異なることに留意してください。この例では、変換は茶色の目を対象とした単純なフィルターで、次のように分離できます。

Copied!1
inputDf = inputDf.filter("eye = 'Brown'");

出力を書き込む

入力データセットの変更を解釈し、入力の必要な部分を読み取り、変換ロジックに従ってデータを加工したら、それに応じて出力を書き込めます。 WriteMode は、さまざまな書き込みモードを提供します。モードは次のとおりです。

SNAPSHOT
UPDATE

たとえば、この例では、入力の変更タイプに基づいて出力タイプを選択できます。

Copied!1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
private WriteMode getWriteMode(FoundryInput input) {
     switch (input.asDataFrame().modificationType()) {
         case UNCHANGED:
             LOG.info("No changes in input dataset, writing in update mode");
             return WriteMode.UPDATE;
         case APPENDED:
             LOG.info("Append-only changes in input dataset, writing in update mode");
             return WriteMode.UPDATE;
         case UPDATED:
             LOG.info("Update-type changes in input dataset, writing in snapshot mode");
             return WriteMode.SNAPSHOT;
         case NEW_VIEW:
             LOG.info("new view in input dataset, writing in snapshot mode");
             return WriteMode.SNAPSHOT;
         default:
             throw new IllegalArgumentException("Unknown ModificationType for input dataset " + input.asDataFrame().modificationType());
     }
 }

WriteMode.UPDATE と DataFrameModificationType.UPDATED は同等ではありません

WriteMode.UPDATE と DataFrameModificationType.UPDATED を混同しないでください。前者は出力データセットをインクリメンタルに変更するため、下流のデータセットでは DataFrameModificationType.APPENDED になります。後者は、データの追加と既存の行の変更の両方を含む入力データセットの変更です。

最後に、書き込みモードを含めるように write() 関数を変更できます。

Copied!1
2
3
4
5
6
7
@Compute
public void myComputeFunction(
        @Input("/examples/students_hair_eye_color") FoundryInput myInput,
        @Output("/examples/students_hair_eye_color_filtered") FoundryOutput myOutput) {
    Dataset<Row> inputDf = myInput.asDataFrame().readForRange(getReadRange(myInput));
    myOutput.getDataFrameWriter(inputDf.filter("eye = 'Brown'")).write(getWriteMode(myInput));
}

すべてを組み合わせる

これらを組み合わせることで、単純なインクリメンタルフィルタリングのトランスフォームを作成できます。

Copied!1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
package myproject.datasets;

import com.palantir.transforms.lang.java.api.*;
import org.apache.spark.sql.Dataset;
import org.apache.spark.sql.Row;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

public final class FilterTransform {

    private static final Logger LOG = LoggerFactory.getLogger(FilterTransform.class);

    @Compute
    public void myComputeFunction(
            @Input("/examples/students_hair_eye_color") FoundryInput myInput,
            @Output("/examples/students_hair_eye_color_filtered") FoundryOutput myOutput) {
        Dataset<Row> inputDf = myInput.asDataFrame().readForRange(getReadRange(myInput));
        myOutput.getDataFrameWriter(inputDf.filter("eye = 'Brown'")).write(getWriteMode(myInput));
    }

   private ReadRange getReadRange(FoundryInput input) {
        switch (input.asDataFrame().modificationType()) {
            case UNCHANGED:
                LOG.info("No changes in input dataset, read only unprocessed");
                return ReadRange.UNPROCESSED;
            case APPENDED:
                LOG.info("Append-only changes in input dataset, read only unprocessed");
                return ReadRange.UNPROCESSED;
            case UPDATED:
                LOG.info("Update-type changes in input dataset, read entire view");
                return ReadRange.ENTIRE_VIEW;
            case NEW_VIEW:
                LOG.info("New view in input dataset, read entire view");
                return ReadRange.ENTIRE_VIEW;
            default:
                throw new IllegalArgumentException("Unknown ModificationType for input dataset "
                    + input.asDataFrame().modificationType());
        }
   }

   private WriteMode getWriteMode(FoundryInput input) {
        switch (input.asDataFrame().modificationType()) {
            case UNCHANGED:
                LOG.info("No changes in input dataset, writing in update mode");
                return WriteMode.UPDATE;
            case APPENDED:
                LOG.info("Append-only changes in input dataset, writing in update mode");
                return WriteMode.UPDATE;
            case UPDATED:
                LOG.info("Update-type changes in input dataset, writing in snapshot mode");
                return WriteMode.SNAPSHOT;
            case NEW_VIEW:
                LOG.info("new view in input dataset, writing in snapshot mode");
                return WriteMode.SNAPSHOT;
            default:
                throw new IllegalArgumentException("Unknown ModificationType for input dataset " + input.asDataFrame().modificationType());
        }
    }
}

上で紹介したように、入力の変更タイプを評価し、それに応じて入力を読み取ります。次に、出力データセットをインクリメンタルに更新するか、新しい SNAPSHOT トランザクションを開始するかを決定します。

ファイルを使用するインクリメンタルな低レベルトランスフォームを記述する

DataFrame ではなく、ReadOnlyLogicalFileSystem と WriteOnlyLogicalFileSystem を通じて生のファイルを読み書きする低レベルトランスフォームにも、上記と同じインクリメンタル処理の原則が適用されます。主な違いは、読み取り範囲と書き込みモードを設定する場所です。Dataset<Row> リーダーと DataFrameWriter ではなく、FoundryFiles と FoundryOutput が返すファイルシステムオブジェクトに設定します。

読み取り範囲と書き込みモードを選択する

FoundryInput.asFiles() は FoundryFiles オブジェクトを返します。その modificationType() メソッドは、DataFrameModificationType に相当するファイルベースの FilesModificationType を返します。変更タイプを使用して、両方の設定を選択します。

`FilesModificationType`	`ReadRange`	`WriteMode`
`UNCHANGED` または `APPENDED`	`UNPROCESSED`	`UPDATE`
`UPDATED` または `NEW_VIEW`	`ENTIRE_VIEW`	`SNAPSHOT`

必要なファイルに範囲を限定した ReadOnlyLogicalFileSystem を取得するには、選択した ReadRange を FoundryFiles.getFileSystem() に渡します。出力データセットを更新または置き換えるには、選択した WriteMode を FoundryOutput.getFileSystem() に渡します。

書き込みモードを省略すると、必ず SNAPSHOT トランザクションになります

引数なしの myOutput.getFileSystem() を呼び出すと、入力にどの ReadRange を使用したかにかかわらず、出力では常に SNAPSHOT トランザクションが発生します。インクリメンタルに書き込むには、WriteMode を明示的に渡してください。

低レベルトランスフォームを組み合わせる

以下の例は、入力データセットから出力データセットにファイルをコピーします。入力がどのように変更されたかに基づいて、読み取り範囲と書き込みモードを選択します。

Copied!1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
package myproject.datasets;

import com.google.common.io.ByteStreams;
import com.palantir.transforms.lang.java.api.Compute;
import com.palantir.transforms.lang.java.api.FoundryFiles;
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.ReadOnlyLogicalFileSystem;
import com.palantir.transforms.lang.java.api.ReadRange;
import com.palantir.transforms.lang.java.api.WriteMode;
import com.palantir.transforms.lang.java.api.WriteOnlyLogicalFileSystem;

public final class IncrementalFileCopyTransform {

    @Compute
    public void myComputeFunction(
            @Input("/examples/raw_files") FoundryInput myInput,
            @Output("/examples/raw_files_copy") FoundryOutput myOutput) {
        FoundryFiles inputFiles = myInput.asFiles();
        ReadRange readRange;
        WriteMode writeMode;
        switch (inputFiles.modificationType()) {
            case UNCHANGED:
            case APPENDED:
                readRange = ReadRange.UNPROCESSED;
                writeMode = WriteMode.UPDATE;
                break;
            case UPDATED:
            case NEW_VIEW:
                readRange = ReadRange.ENTIRE_VIEW;
                writeMode = WriteMode.SNAPSHOT;
                break;
            default:
                throw new IllegalArgumentException("Unknown FilesModificationType for input dataset "
                    + inputFiles.modificationType());
        }

        ReadOnlyLogicalFileSystem inputFileSystem = inputFiles.getFileSystem(readRange);
        WriteOnlyLogicalFileSystem outputFileSystem = myOutput.getFileSystem(writeMode);

        inputFileSystem.listAllFiles().forEach(path -> inputFileSystem.readFrom(path, inputStream -> {
            outputFileSystem.writeTo(path, outputStream -> ByteStreams.copy(inputStream, outputStream));
            return null;
        }));
    }
}

このトランスフォームは、必要な入力ファイルだけを読み取り、出力に対してインクリメンタルな更新を行うか、完全なスナップショットを作成するかを選択します。インクリメンタルな動作を使用せずに ReadOnlyLogicalFileSystem と WriteOnlyLogicalFileSystem でファイルを読み書きするその他の例については、非構造化ファイルを読み書きするを参照してください。

ベストプラクティス

スナップショットとインクリメンタルを切り替える

通常はインクリメンタル（差分処理）トランスフォームを実行しつつ、ときどきデータセットのスナップショットを再実行する必要がある場合を考えます。

望ましい結果を手動でハードコードすることを避けるために、新しい入力を追加し、その入力が変更されるたびに出力で SNAPSHOT 書き込みモードが使用されるようにできます。この新しい入力は実質的にスナップショットをトリガーするデータセットとして機能します。このスナップショットトリガーデータセットの変更タイプに基づいて、トランスフォームのほかの入力の読み取り範囲を調整する必要があります。

来歴情報のない空の APPEND トランザクションを作成することで、外部からスナップショットを強制することもできます。ただし、transforms-java はこの機能を公開していないため、このガイドの対象範囲外です。

高度な機能

このセクションの高度な機能は、誤って使用すると重大な悪影響を及ぼす可能性があります。影響を完全に理解していない場合は、これらの機能を使用_しないでください_。適切な注意を払わずに実行すると、意図しない結果が生じるリスクが高くなります。ご不明な点がある場合は、Palantir の担当者にお問い合わせください。

高度な機能は通常、@Compute 関数の上に注釈を付けることで組み込まれます。トランスフォームで手動登録を使用している場合は、代わりに Transform Builder にプロパティを追加します。

インクリメンタル削除を無視する

インクリメンタルビルドが、際限なく増加する追加専用データセットに依存しており、その増加に対応するディスク容量が不足している場合は、上流データセットの一部を削除する必要が生じることがあります。ただし、元のデータセットを変更すると変更タイプが APPENDED にならないため、インクリメンタル処理を維持できなくなる可能性があります。 IncrementalOptions.IGNORE_INCREMENTAL_DELETES を使用すると、これを回避し、上流データセットでの削除を破壊的変更として扱わなくなります。

インクリメンタル削除を無視できるのは、低レベルトランスフォームのみです。

Copied!1
2
3
4
5
6
    @Compute
    @UseIncrementalOptions(IncrementalOptions.IGNORE_INCREMENTAL_DELETES)
    public void myComputeFunction(
           @Input("/Users/admin/students_data") FoundryInput myInput,
           @Output("/Users/admin/students_data_filtered") FoundryOutput myOutput) {
    ...

トランスフォームを手動で登録している場合は、次のコードブロックのようにビルダーにプロパティを追加します。

Copied!1
2
3
4
5
6
    LowLevelTransform lowLevelManualTransform = LowLevelTransform.builder()
               .computeFunctionInstance(new MyLowLevelManualFunction())
               .putParameterToInputAlias("myInput", "/path/to/input/dataset")
               .putParameterToOutputAlias("myOutput", "/path/to/output/dataset")
               .ignoreIncrementalDeletes(true)
               .build();

スキーマ変更を無視する

入力データセットのスキーマ変更は、インクリメンタルトランスフォームと組み合わせた場合に予期しない結果をもたらす可能性があります。この機能を使用する前に、このセクションの残りを読み、考えられるすべての影響を理解していることを確認してください。

スキーマ変更を無視できるのは、低レベルトランスフォームのみです。

インクリメンタルビルドが依存するデータセットのスキーマが変更されると、その変更によって DataFrameModificationType.NEW_VIEW が発生し、インクリメンタル処理を維持できなくなる可能性があります。

ただし、IncrementalOptions.USE_SCHEMA_MODIFICATION_TYPE オプションが設定されている場合、スキーマ変更によって新しいビューが発生することはありません。代わりに、入力データセットのスキーマ変更は DataFrameModificationType.UNCHANGED として解釈され、SchemaModificationType フラグ SchemaModificationType.NEW_SCHEMA が設定されます。これにより、ユーザーはこの特殊なケースを明示的に処理できます。

Copied!1
2
3
4
5
6
7
        @Compute
        @UseIncrementalOptions(IncrementalOptions.USE_SCHEMA_MODIFICATION_TYPE)
        public void myComputeFunction(
               @Input("/Users/admin/students_data") FoundryInput myInput,
               @Output("/Users/admin/students_data_filtered") FoundryOutput myOutput) {
        ...
        }

トランスフォームを手動で登録している場合は、次のコードブロックのようにビルダーにプロパティを追加します。

Copied!1
2
3
4
5
6
    LowLevelTransform lowLevelManualTransform = LowLevelTransform.builder()
               .computeFunctionInstance(new MyLowLevelManualFunction())
               .putParameterToInputAlias("myInput", "/path/to/input/dataset")
               .putParameterToOutputAlias("myOutput", "/path/to/output/dataset")
               .useSchemaModificationType(true)
               .build();

トランスフォームに関連するビルドが成功するか失敗するかは、トランスフォームが入力データセットにどのように依存しているかによって決まります。より具体的には、トランスフォームがスキーマ変更の対象となる列に依存している場合、その列を変更するとインクリメンタルトランスフォームが失敗します。このような場合、インクリメンタルトランスフォームを再び使用するには、その前に新しいスナップショットが必要です。

次の場合、トランスフォームは特定の列に依存しています。

その列に明示的に依存する変更が含まれている場合（たとえば、filter("eye = 'Brown'") があり、生データセット内の列 eye の名前が変更されたり、列が削除されたりすると、FilterTransform を再トリガーしたときにインクリメンタル更新が失敗します）。
変更された列が出力データセットに含まれていた場合（たとえば、この例の生データセットから列 hair を削除すると、FilterTransform が失敗します）。

トランスフォームがスキーマ変更に依存していない場合、インクリメンタルビルドは成功します。

たとえば、最初にトランスフォームに id と eye の select 文を追加し、生データセットから初回のスナップショットビルドをトリガーしてから、生データセットの列 hair を削除した場合、インクリメンタルビルドは成功し、スキーマ変更はインクリメンタルトランスフォームに影響を及ぼしません。また、スキーマへの追加変更（たとえば、新しい列の追加）の場合も、ビルドは常に成功します。

←

PREVIOUS基本的なトランスフォーム / 詳細設定

NEXTユーザー定義関数

→