メディアセットを使用するインクリメンタル（差分処理）トランスフォームを記述する

メディアセットは、インクリメンタルに読み取りおよび書き込みができます。インクリメンタルトランスフォームの概要と使用するタイミングについては、インクリメンタルの概要とインクリメンタルのリファレンスを参照してください。

Copied!
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
from transforms.api import transform, incremental
from transforms.mediasets import MediaSetInput, MediaSetOutput

@incremental(v2_semantics=True)
@transform(
    input_PNGs=MediaSetInput('/examples/input_PNGs'),
    output_PNGs=MediaSetOutput('/examples/output_PNGs'),
)
def upload_pngs(input_PNGs, output_PNGs):

    # 前回のビルド以降に追加されたメディアアイテムのみを含むデータフレームを返します
    listed_pngs = input_PNGs.dataframe()

    def fast_copy_media_item(row):
        output_PNGs.fast_copy_media_item(input_PNGs, row.media_item_rid, row.path)

    # `listed_pngs` 内のすべての項目を出力メディアセットに高速コピーします
    # このトランスフォームがインクリメンタルに実行されている場合、これらの項目は出力に追加されます。
    # インクリメンタルに実行されていない場合は、これらの項目で出力が置き換えられます
    listed_pngs.foreach(fast_copy_media_item)

上の例では、トランスフォームは modify 書き込みモードを使用して output_PNGs に書き込みます。前回のビルド以降に入力メディアセットに追加されたメディアアイテムのみが処理されます。トランスフォームをインクリメンタルに実行できない場合、出力は replace 書き込みモードで書き込まれ、入力全体が読み取られます。要件については以下を参照してください。

v2_semantics が True に設定されている場合、インクリメンタルメディアセットは、任意の数のほかのインクリメンタル入力および出力と組み合わせて使用できます。これにはデータセットと仮想テーブルが含まれます。

インクリメンタル計算の要件

すべてのインクリメンタル入力および出力が、トランスフォームをインクリメンタルに実行できるかどうかの判定に関与します。データセットによってトランスフォームをインクリメンタルに実行できなくなる条件については、インクリメンタルトランスフォームのリファレンスを参照してください。

メディアセット出力によって、次の場合にトランスフォームをインクリメンタルに実行できなくなることがあります。

• 複数出力のビルドで、ほかの出力とは異なるトランスフォームで最後にビルドされた場合。
• トランザクション型メディアセットであり、直近のビルド以降に変更された場合。これにはユーザーによるアップロードと削除が含まれます。

メディアセット入力によって、次の場合にトランスフォームをインクリメンタルに実行できなくなることがあります。

• メディアセットの内容が置き換えられた場合。たとえば、replace 書き込みモードを使用して書き込まれた場合です。

メディアセット入力が snapshot_input として含まれている場合、内容が置き換えられても、ビルドのインクリメンタル実行を妨げません。スナップショット入力に関するドキュメントを参照してください。

データセットとは異なり、パスの上書きとメディアアイテムの削除によって、トランスフォームをインクリメンタルに実行できなくなることは_ありません_。

インクリメンタル読み取りモード

インクリメンタルトランスフォームでは、次の3つのモードのいずれかを使用して、メディアセット入力の一覧を取得できます。

• added： 前回のビルド以降にブランチに追加された項目のみが含まれます。
• previous： 前回のビルドが実行された時点でブランチに存在していた項目のみが含まれます。
• current： メディアセットのブランチにあるすべての項目が含まれます。

added と previous の和集合は、常に current と等しくなります。

デフォルトの読み取りモードは、インクリメンタルに実行する場合は added、そうでない場合は current です。ただし、どの一覧取得メソッドでも mode パラメーターを使用して読み取りモードを指定できます。

Copied!
1
2
3
4
5
6
7
8
9
10
11
12
13
14
from transforms.api import transform, incremental
from transforms.mediasets import MediaSetInput, MediaSetOutput

@incremental(v2_semantics=True)
@transform(
    input_PNGs=MediaSetInput('/examples/input_PNGs'),
    output_PNGs=MediaSetOutput('/examples/output_PNGs'),
)
def upload_pngs(input_PNGs, output_PNGs):
    # インクリメンタルに実行されている場合は `added`、そうでない場合は `current` を使用します
    listed_pngs = input_PNGs.dataframe(deduplicate_by_path=False)

    # 常に `previous` モードで読み込みます
    previous_listed_pngs = input_PNGs.dataframe(deduplicate_by_path=False, mode="previous")

パスが上書きされ、一覧取得時にパスによる重複排除が行われる場合、最新の項目のみが含まれます。特定のパスにあるすべての入力項目を処理するには、常に deduplicate_by_path=False を指定する必要があります。

インクリメンタル書き込みモード

インクリメンタルメディアセット出力への書き込み時には、実行時に書き込みモードを設定できます。これは、ビルドをインクリメンタルに実行するかどうかを決めるカスタムロジックがトランスフォームに含まれている場合に役立ちます。以下の例では、前回のビルド以降にいずれかのパスが上書きされていた場合、ビルドはインクリメンタルに実行されません。

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
from transforms.api import transform, incremental
from transforms.mediasets import MediaSetInput, MediaSetOutput

@incremental(v2_semantics=True)
@transform(
    input_PNGs=MediaSetInput('/examples/input_PNGs'),
    output_PNGs=MediaSetOutput('/examples/output_PNGs'),
)
def upload_pngs(input_PNGs, output_PNGs):
    previous_dataframe = input_PNGs.dataframe(deduplicate_by_path=False, mode="previous")
    added_dataframe = input_PNGs.dataframe(deduplicate_by_path=False, mode="added")

    # `input_PNGs` メディアセット内でパスが上書きされたかどうかを、
    # このトランスフォームの直近の実行以降について判定します
    paths_overwritten = previous_dataframe.join(added_dataframe, mode="inner", on="path").count() > 0

    if paths_overwritten:
        # 入力メディアセット全体が読み込まれ、出力メディアセットは
        # このトランスフォームで書き込まれた項目で置き換えられます
        read_mode = "current"
        output_PNGs.set_write_mode("replace")
    else:
        # 入力メディアセット内の新しく追加された項目のみが読み込まれ、このトランスフォームで書き込まれた項目が
        # 出力メディアセットに追加されます
        read_mode = "added"
        output_PNGs.set_write_mode("modify")

インクリメンタルトランスフォームとブランチ

メディアセットはインクリメンタルのフォールバックブランチをサポートしていません。新しいブランチでインクリメンタルトランスフォームを実行すると、出力が現在空であるため、インクリメンタルデコレーターは SNAPSHOT を推奨します。そのため、メインブランチで同じビルドを実行しても、必ずしも SNAPSHOT になるとは限りません。

インクリメンタルトランスフォームとトランザクションレスメディアセット

トランザクションレスメディアセットは modify 書き込みモードを使用し、replace 書き込みモードは使用できません。つまり、トランザクションレスメディアセットは SNAPSHOT にできません。トランザクションレスメディアセットがインクリメンタルトランスフォームの出力であり、そのトランスフォームをインクリメンタルに実行できない場合、ビルドは失敗します。この場合は、ビルドをインクリメンタルに実行できない理由を調査する必要があります。

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

ビルド中に個々のメディアセット出力を中止することはできません。個々の出力を中止する代わりに、TransformContext の .abort_job() メソッドを使用して、ジョブ全体を中止することを推奨します。これにより、後続の実行をインクリメンタルにできます。

インクリメンタル入力のバッチサイズを制限する

通常、インクリメンタルトランスフォームでは、各入力メディアセットの未処理のメディアアイテムはすべて、同じジョブで処理されます。ただし、状況によっては、ジョブで処理されるメディアアイテムの数が大きく変わることがあります。

• インクリメンタルトランスフォームが SNAPSHOT モードでビルドされ、入力全体が最初から読み取られる場合。たとえば、トランスフォームのセマンティックバージョンが引き上げられた場合です。
• 入力メディアセットに多数のメディアアイテムが蓄積され、その結果、下流のインクリメンタルトランスフォームが1つのジョブで多数のメディアアイテムを処理する必要がある場合。

トランスフォームの各インクリメンタル入力メディアセットにバッチ制限を設定すると、各ジョブで読み取るメディアアイテムの数を制限できます。この概念は、入力データセットのトランザクション制限に類似しています。

バッチ制限は、ジョブごとに処理されるメディアアイテム数の上限です。この制限は、メディアアイテムのパスによる重複排除が行われる前に適用されます。たとえば、バッチ制限が100で、トランスフォームがメディアアイテムのパスによる重複排除を行う場合、処理されたメディアアイテムが100個未満に見えることがあります。

以下の例では、次を使用するようにインクリメンタルトランスフォームを設定します。

• それぞれ異なるバッチ制限を持つ2個のインクリメンタルメディアセット入力。
• バッチ制限を使用しないインクリメンタルメディアセット入力。

Copied!
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
from transforms.api import incremental, transform
from transforms.mediasets import MediaSetInput, MediaSetOutput

@incremental(
    v2_semantics=True,
    strict_append=True,
)
@transform(
    input_media_set_1=MediaSetInput("/examples/media-set-input_1", batch_limit=100),
    input_media_set_2=MediaSetInput("/examples/media-set-input_2", batch_limit=500),
    input_media_set_3=MediaSetInput("/examples/media-set-input_3"),
    output_media_set=MediaSetOutput("/examples/media-set-output"),
)
def compute(ctx, input_media_set_1, input_media_set_2, input_media_set_3, output_media_set):
    ...

想定される使用方法は、出力を最新の状態に保持するためにビルドスケジュールを作成することです。