スナップショットから履歴データセットを作成する

可能であれば、履歴データセットは APPEND トランザクションとして取り込むことがベストプラクティスです。詳細については、以下の警告を参照してください。

ワークフローの概要

日、週、または時間ごとに、新しい SNAPSHOT インポートによって以前のビューがデータセットの現在のデータに置き換えられる生データセットを扱うことがあります。ただし、以前のビューから何が変わったかを確認するには、以前のデータを利用できると便利なことがよくあります。前述のとおり、APPEND トランザクションを使用し、インポート日を格納する列を追加して、取り込みの一環としてこの処理を行うことがベストプラクティスです。ただし、それができない場合は、Python トランスフォームの incremental デコレーターを使用して、これらの定期的なスナップショットをそのデータセットの履歴バージョンに追加できます。この方法の注意点と潜在的な脆弱性については、以下の警告を参照してください。

履歴データセットの例。

サンプルコード

以下に、incremental デコレーターを使用して履歴データセットを作成する例を示します。

Copied!1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
from datetime import date

import polars as pl
from transforms.api import transform, incremental, Input, Output

@incremental(snapshot_inputs=['input_data'])
@transform.using(
    input_data=Input("/path/to/snapshot/input"),
    history=Output("/path/to/historical/dataset"),
)
def my_compute_function(input_data, history):
    input_df = input_data.polars()

    # 入力が1日に2回以上変更される可能性がある場合は、代わりに datetime.now() をインポートして使用します。
    input_df = input_df.with_columns(pl.lit(date.today()).alias('date'))

    history.write_table(input_df)

Copied!1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
from transforms.api import transform, incremental, Input, Output

@incremental(snapshot_inputs=['input_data'])
@transform.using(
    input_data=Input("/path/to/snapshot/input"),
    history=Output("/path/to/historical/dataset"),
)
def my_compute_function(ctx, input_data, history):
    duckdb = ctx.duckdb()

    # 入力が1日に2回以上変更される可能性がある場合は、代わりに CURRENT_TIMESTAMP を使用します。
    query = duckdb.conn.sql(f"""
        SELECT *, CURRENT_DATE AS date
        FROM input_data
    """)

    history.write_table(query)

Copied!1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
import pandas as pd
from transforms.api import transform, incremental, Input, Output

@incremental(snapshot_inputs=['input_data'])
@transform.using(
    input_data=Input("/path/to/snapshot/input"),
    history=Output("/path/to/historical/dataset"),
)
def my_compute_function(input_data, history):
    input_df = input_data.pandas()

    # 入力が1日に2回以上変更される可能性がある場合は、.date() を省略します。
    input_df['date'] = pd.Timestamp.now().date()

    history.write_table(input_df)

Copied!1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
from transforms.api import transform, incremental, Input, Output
from pyspark.sql.functions import current_date

@incremental(snapshot_inputs=['input_data'])
@transform.spark.using(
    input_data=Input("/path/to/snapshot/input"),
    history=Output("/path/to/historical/dataset"),
)
def my_compute_function(input_data, history):
    input_df = input_data.dataframe()

    # 入力が1日に2回以上変更される可能性がある場合は、代わりに current_timestamp() をインポートして使用します。
    input_df = input_df.withColumn('date', current_date())

    history.write_dataframe(input_df)

この方法が機能する理由

インクリメンタル（差分処理）デコレーターは、入力と出力の読み取り/書き込みモードに追加のロジックを適用します。上記の例では、入力と出力にデフォルトの読み取り/書き込みモードを使用しています。

読み取りモード

SNAPSHOT 入力を使用する場合、デフォルトの読み取りモードは current です。これは、前回のトランザクション以降に追加された行だけではなく、入力 DataFrame 全体を取得することを意味します。入力データセットが APPEND トランザクションから作成されている場合は、incremental デコレーターを使用して added 読み取りモードを使うことで、前回のビルド以降に追加された行のみにアクセスできます。トランスフォームは current 出力からスキーマ情報を取得するため、DataFrame の previous バージョンを読み取る場合のように、スキーマ情報を渡す必要はありません。そのような場合の例は dataframe('previous', schema=input.schema) です。

書き込みモード

トランスフォームがインクリメンタルに実行されるとは、出力のデフォルトの書き込みモードが modify に設定されることを意味します。このモードでは、ビルド中に書き込まれたデータで既存の出力を変更します。たとえば、出力が modify モードのときに write_dataframe() を呼び出すと、書き込まれた DataFrame が既存の出力に追加されます。このケースでは、まさにこの処理が行われています。

データセットにデータが蓄積されることを確認する

データの蓄積は、1回のビルド内ではなく、連続するビルド間で行われます。最初のビルドでは、現在の入力スナップショットを使用して履歴が作成されます。入力に新しい SNAPSHOT トランザクションが到着すると、次のビルドでその時点のスナップショットが既存の履歴に追加されます。

履歴が蓄積されることを確認するには、次の手順に従います。

トランスフォームを1回実行します。一度もビルドされていない出力は非インクリメンタルに実行され、replace 書き込みモードを使用するため、最初のビルド後にスナップショットが1つだけ表示されるのは期待される動作です。
たとえばソースデータを再インポートまたは再同期して、入力データセットに新しい SNAPSHOT トランザクションを到着させてから、トランスフォームを再度実行します。新しい入力トランザクションがないビルドでは、新しい履歴状態は追加されません。
出力を調べ、例で追加した日付またはタイムスタンプの列によって識別される、両方のスナップショットの行が含まれていることを確認します。スナップショットが1日に2回以上到着する可能性がある場合は、日付ではなくタイムスタンプを使用します。
ビルドがインクリメンタルに実行されたかどうかを確認するには、そのジョブのインクリメンタルレポートを確認します。
- データセットプレビュー、Data Lineage、または Builds アプリケーションからビルドレポートを表示します。
- ガントチャートの下にあるリストでジョブを選択し、Spark の詳細を選択して、スナップショット/インクリメンタルタブを開きます。レポートには、ジョブが各入力について読み取ったトランザクションの範囲が表示されます。

プレビューは非インクリメンタルに実行されます

Code Repositories のプレビュー機能は、require_incremental=True が設定されている場合でも、常にトランスフォームを非インクリメンタルに実行します。そのため、実行のたびに出力が現在のスナップショットで置き換えられるので、Code Repositories のプレビューではデータの蓄積を確認できません。代わりに、トランスフォームをインクリメンタルに実行する VS Code のプレビューを使用するか、スケジュールされたビルドまたは手動ビルドの出力を確認してください。

異なる入力トランザクションで2回ビルドした後も、出力に最新のスナップショットしか含まれていない場合、2回目のビルドが非インクリメンタルに実行され、出力が置き換えられた可能性があります。よくある原因は次のとおりです。

semantic_version の変更
全体が書き換えられる入力が、スナップショット入力として宣言されていない
出力の最後のトランザクションが別のトランスフォームによって生成されている

原因を確認するには、インクリメンタル計算の要件を確認し、上記の手順に従ってビルドのジョブのインクリメンタルレポートを確認します。ジョブのドライバーログにも、警告 transforms.api._incremental: Not running incrementally が記録されます。

警告

このトランスフォームは SNAPSHOT データセットを入力として使用するため、ビルドの失敗やその他の理由でビルドが取り逃した可能性のあるスナップショットを復元する方法はありません。これが懸念される場合は、この方法を使用しないでください。代わりに、入力データセットの所有者に連絡し、データセットの以前のトランザクションにアクセスできるように、APPEND データセットに変換できるかどうかを確認してください。インクリメンタル計算は、このように動作するように設計されています。

この方法は、次の条件で失敗します。

入力データセットに列が追加される
既存のテーブルの列数が入力データのスキーマと一致しない
入力データセットの列のデータ型が変わる（たとえば integer から decimal へ）
入力データセットを変更する（そのデータセットのスキーマが同一であっても）。この場合、追加するのではなく、出力が入力で完全に置き換えられます。

リソース消費の増加

このパターンを使用すると、履歴データセットに小さなファイルが蓄積される可能性があります。ファイルの蓄積は望ましい結果ではなく、この履歴データセットを使用する下流のトランスフォームや分析におけるリソース消費やビルド時間の増加につながります。各ファイルの読み取りにはオーバーヘッドが伴うため、バッチ計算と対話型計算の時間が長くなる可能性があります。圧縮はデータセット内のファイル間ではなく、ファイルごとに行われるため、ディスク使用量が増加する可能性があります。定期的にデータのスナップショットを再作成するロジックを構築して、この動作を防ぐことができます。

出力ファイルの数を調べることで、最適なインクリメンタル書き込みモードを特定できます。このモードでは、前回のトランザクションの出力を DataFrame として読み込み、それを受信データと和集合で結合し、データファイルをまとめることができます。これにより、多数の小さなファイルが1つの大きなファイルになります。

次の例のように、出力データセットのファイルシステム内のファイル数を調べ、if 文を使用して write_mode を設定します。

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
import polars as pl
from transforms.api import transform, Input, Output, incremental

FILE_COUNT_LIMIT = 100

# 必ずここに必要な出力スキーマを挿入してください
schema = {
    'Value': pl.Float64,
    'Time': pl.Datetime,
    'DataGroup': pl.Utf8
}

def compute_logic(df):
    """
    これがトランスフォームのロジックです
    """
    return df.filter(pl.lit(True))

@incremental(semantic_version=1)
@transform.using(
    output=Output("/Org/Project/Folder1/Output_dataset_incremental"),
    input_df=Input("/Org/Project/Folder1/Input_dataset_incremental"),
)
def compute(input_df, output):
    df = input_df.polars('added')
    df = compute_logic(df)

    # データセット内のファイルの一覧を取得します
    files = list(output.filesystem(mode='previous').ls())

    if (len(files) > FILE_COUNT_LIMIT):
        # インクリメンタル（差分処理）でマージして置き換える
        previous_df = output.polars('previous', schema)
        df = pl.concat([df, previous_df])
        mode = 'replace'
    else:
        # 標準のインクリメンタルモード
        mode = 'modify'

    output.set_mode(mode)
    output.write_table(df)

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
import polars as pl
from transforms.api import transform, Input, Output, incremental

FILE_COUNT_LIMIT = 100

# 必ずここに必要な出力スキーマを挿入してください
schema = {
    'Value': pl.Float64,
    'Time': pl.Datetime,
    'DataGroup': pl.Utf8
}

def compute_logic(conn):
    """
    これがトランスフォームのロジックです
    """
    return conn.sql("SELECT * FROM input_df")

@incremental(semantic_version=1)
@transform.using(
    output=Output("/Org/Project/Folder1/Output_dataset_incremental"),
    input_df=Input("/Org/Project/Folder1/Input_dataset_incremental"),
)
def compute(ctx, input_df, output):
    conn = ctx.duckdb().conn
    df = compute_logic(conn).pl()

    # データセット内のファイルの一覧を取得します
    files = list(output.filesystem(mode='previous').ls())

    if (len(files) > FILE_COUNT_LIMIT):
        # インクリメンタル（差分処理）でマージして置き換える
        previous_df = output.polars('previous', schema)
        df = pl.concat([df, previous_df])
        mode = 'replace'
    else:
        # 標準のインクリメンタルモード
        mode = 'modify'

    output.set_mode(mode)
    output.write_table(df)

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
import pandas as pd
from transforms.api import transform, Input, Output, incremental

FILE_COUNT_LIMIT = 100

# 必ずここに必要な出力スキーマを挿入してください
schema = {
    'Value': 'float64',
    'Time': 'datetime64[ns]',
    'DataGroup': 'object'
}

def compute_logic(df):
    """
    これがトランスフォームのロジックです
    """
    return df[df.index >= 0]  # filter(True) と同等です

@incremental()
@transform.using(
    output=Output("/Org/Project/Folder1/Output_dataset_incremental"),
    input_df=Input("/Org/Project/Folder1/Input_dataset_incremental"),
)
def compute(input_df, output):
    df = input_df.pandas('added')
    df = compute_logic(df)

    # データセット内のファイルの一覧を取得します
    files = list(output.filesystem(mode='previous').ls())

    if (len(files) > FILE_COUNT_LIMIT):
        # インクリメンタル（差分処理）でマージして置き換える
        previous_df = output.pandas('previous', schema)
        df = pd.concat([df, previous_df], ignore_index=True)
        mode = 'replace'
    else:
        # 標準のインクリメンタルモード
        mode = 'modify'

    output.set_mode(mode)
    output.write_table(df)

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
from pyspark.sql import types as T
from transforms.api import transform, Input, Output, incremental

FILE_COUNT_LIMIT = 100

# 必ずここに必要な出力スキーマを挿入してください
schema = T.StructType([
    T.StructField('Value', T.DoubleType()),
    T.StructField('Time', T.TimestampType()),
    T.StructField('DataGroup', T.StringType())
])


def compute_logic(df):
    """
    これがトランスフォームのロジックです
    """
    return df.filter(True)

@incremental()
@transform.spark.using(
    output=Output("/Org/Project/Folder1/Output_dataset_incremental"),
    input_df=Input("/Org/Project/Folder1/Input_dataset_incremental"),
)
def compute(input, output):
    df = input.dataframe('added')
    df = compute_logic(df)

    # データセット内のファイルの一覧を取得します
    files = list(output.filesystem(mode='previous').ls())

    if (len(files) > FILE_COUNT_LIMIT):
        # インクリメンタル（差分処理）でマージして置き換える
        previous_df = output.dataframe('previous', schema)
        df = df.unionByName(previous_df)
        mode = 'replace'
    else:
        # 標準のインクリメンタルモード
        mode = 'modify'

    output.set_mode(mode)
    output.write_dataframe(df.coalesce(1))