パイプライン

リポジトリ内の各 Python トランスフォームサブプロジェクトは、単一の transforms.api.Pipeline オブジェクトを公開します。この Pipeline オブジェクトは、次の目的で使用されます。

1. データセットをビルドするための指示とともに、Foundry にデータセットを登録します。
2. Foundry のビルド中に、指定されたデータセットのビルドを担当する @transform ロジックを見つけて実行します。

ほとんどの場合、デフォルトのリポジトリ設定でトランスフォームが自動的に登録されるため、Pipeline オブジェクトに特別な設定は必要ありません。

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

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

自動登録

プロジェクトが複雑になるにつれて、Pipeline オブジェクトにトランスフォームを手動で追加する作業は煩雑になることがあります。これを解消するため、Pipeline オブジェクトには、Python モジュールまたはパッケージ内のすべてのトランスフォームを再帰的に見つける discover_transforms() メソッドが用意されています。

Copied!
1
2
3
4
5
6
from transforms.api import Pipeline
import my_module  # このモジュールにトランスフォームの定義があります


my_pipeline = Pipeline()
my_pipeline.discover_transforms(my_module)

手動登録

add_transforms() 関数を使用して、Pipeline オブジェクトにトランスフォームを手動で追加できます。この関数は、任意の数のトランスフォームを受け取り、パイプラインに追加します。また、同じ出力データセットを宣言するトランスフォームが2個存在しないかどうかも確認します。

Copied!
1
2
3
4
5
6
7
8
9
10
11
12
from transforms.api import transform, Pipeline, Input, Output


@transform.using(
    my_output=Output('/path/to/output/dataset'),
    my_input=Input('/path/to/input/dataset')
)
def my_compute_function(my_output, my_input):
    my_output.write_table(my_input.polars())

my_pipeline = Pipeline()
my_pipeline.add_transforms(my_compute_function)

トランスフォームの生成

複数の出力を作成するデータ変換を定義する場合は、トランスフォームの生成を使用するか、複数出力のトランスフォームを定義できます。トランスフォームの生成では、出力ごとに同じ入力を読み込んで処理する必要がある場合があります。複数出力のトランスフォームでは、入力の読み込みと処理を1回だけ行うことができます。詳細については、複数出力のトランスフォームの最適化に関するドキュメントを確認してください。

同じデータ変換ロジックを複数のトランスフォームで再利用したい場合があります。たとえば、次のシナリオを考えてみます。

• さまざまな州に関する情報を含む入力データセットがあります。入力を州で絞り込み、さまざまな統計値を計算するコードがあります。
• null 値を含む可能性がある入力データセットが複数あります。すべての null 値を削除するコードがあります。

どちらの場合も、同じデータ変換コードを複数のトランスフォームで使用すると便利です。出力ごとにトランスフォームオブジェクトを個別に定義する代わりに、for ループを使用してトランスフォームオブジェクトを生成し、プロジェクトのパイプラインに一括登録できます。

以下は、トランスフォームジェネレーターの例です。

Copied!
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
import polars as pl
from transforms.api import transform, Input, Output


def transform_generator(sources):
    transforms = []
    # この例では、複数の入力データセットを使用します。また、複数の出力を
    # 単一の入力データセットから生成することもできます。
    for source in sources:
        @transform.using(
            my_input=Input('/sources/{source}/input'.format(source=source)),
            output=Output('/sources/{source}/output'.format(source=source))
        )
        def compute_function(my_input, output, source=source):
            # 関数内で source 変数をキャプチャするには、デフォルト値を持つキーワード引数として渡します。
            df = my_input.polars()
            filtered = df.filter(pl.col('source') == source)
            output.write_table(filtered)
        transforms.append(compute_function)
    return transforms

TRANSFORMS = transform_generator(['src1', 'src2', 'src3'])

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, Input, Output

def transform_generator(sources):
    transforms = []
    # この例では、複数の入力データセットを使用します。また、複数の出力を
    # 単一の入力データセットから生成することもできます。
    for source in sources:
        @transform.using(
            my_input=Input('/sources/{source}/input'.format(source=source)),
            output=Output('/sources/{source}/output'.format(source=source))
        )
        def compute_function(ctx, my_input, output, source=source):
            # 関数内で source 変数をキャプチャするには、デフォルト値を持つキーワード引数として渡します。
            conn = ctx.duckdb().conn
            query = conn.sql(f"""SELECT * FROM my_input WHERE source = '{source}'""")
            output.write_table(query)
        transforms.append(compute_function)
    return transforms

TRANSFORMS = transform_generator(['src1', 'src2', 'src3'])

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, Input, Output

def transform_generator(sources):
    transforms = []
    # この例では、複数の入力データセットを使用します。また、複数の出力を
    # 単一の入力データセットから生成することもできます。
    for source in sources:
        @transform.using(
            my_input=Input('/sources/{source}/input'.format(source=source)),
            output=Output('/sources/{source}/output'.format(source=source))
        )
        def compute_function(my_input, output, source=source):
            # 関数内で source 変数をキャプチャするには、デフォルト値を持つキーワード引数として渡します。
            df = my_input.pandas()
            filtered = df[df['source'] == source]
            output.write_table(filtered)
        transforms.append(compute_function)
    return transforms

TRANSFORMS = transform_generator(['src1', 'src2', 'src3'])

手動登録を使用する場合は、生成したトランスフォームをパイプラインに追加できます。* 構文に詳しくない場合は、Python ドキュメント ↗を参照してください。

Copied!
1
2
3
4
import my_module

my_pipeline = Pipeline()
my_pipeline.add_transforms(*my_module.TRANSFORMS)

トランスフォームを生成する際は、エラーやビルドの失敗を避けるため、次の点に留意してください。

• ソース値のキャプチャ： 関数内でソース変数をキャプチャするには、compute 関数でデフォルト値を持つ source キーワード引数を渡す必要があります。
• for ループによるトランスフォームの生成： Python の for ループは新しいスコープを作成しないため、トランスフォームオブジェクトを生成するループは関数内に配置する必要があります。関数を使用しない場合、自動登録では for ループ内で定義された最後のトランスフォームオブジェクトだけが誤って検出されます。この関数は、生成されたトランスフォームオブジェクトのリストを返し、その戻り値を変数に代入する必要があります。自動登録で検出されるように設定されたモジュール内でこれらの条件を満たすことで、生成したトランスフォームを自動登録できます。あるいは、手動登録を使用することもできます。
• 入力データセットのリストの変更： ビルド間で入力データセットのリストが変更されると、ビルドに失敗します。たとえば、ビルド間で変更されるファイルから入力データセットのリストを読み込む場合が該当します。これは、ビルドの Job Spec に新しいデータセットの参照が見つからないためです。
• 入力と出力の動的な命名： トランスフォームでは、入力と出力の動的な命名はできません。CI ジョブの実行時に、一意の識別子とデータセット名のリンクを含め、入力と出力の間のすべての関係が決定されます。存在しない出力データセットが作成され、それらに Job Spec が追加されます。
  • データセットをビルドする際は、リポジトリ、ソースファイル、およびデータセットを作成する関数のエントリーポイントへの参照が Job Spec から取得されます。その後、ビルドプロセスが開始され、関数が呼び出されて最終結果が生成されます。入力または出力に変更がある状態でビルドプロセスが開始されると、Job Spec が有効でなくなっているため、エラーが発生します。これにより、一意の識別子とデータセット名の接続が途切れます。
• Code Repositories での手動登録： Code Repositories の Build ボタンは、手動登録では機能しない場合があり、要求されたファイルからパイプライン内のトランスフォームが見つかりませんでしたというエラーが表示されます。その場合でも、Data Lineage またはデータセットプレビューでこれらのデータセットをビルドできます。

エントリーポイント

ほとんどのユースケースでは、デフォルトのエントリーポイントと Pipeline オブジェクトの設定で十分です。エントリーポイントの設定は、複数のパイプラインを持つリポジトリや、異なるディレクトリ構造を使用する場合にのみ推奨します。

Python トランスフォームの実行を担当するランタイムは、プロジェクトの Pipeline オブジェクトを見つけられる必要があります。Pipeline オブジェクトをエクスポートするには、Python トランスフォームサブプロジェクトの setup.py ファイルにある entry_points 引数に追加します。エントリーポイントの詳細については、setuptools ライブラリのドキュメント ↗を参照してください。

デフォルトでは、各 Python サブプロジェクトは、root という名前の transforms.pipelines エントリーポイントをエクスポートする必要があります。このエントリーポイントは、モジュール名と Pipeline 属性を参照します。

たとえば、以下のように「my_pipeline」という名前の Pipeline が myproject/pipeline.py で定義されているとします。

Copied!
1
2
3
4
from transforms.api import Pipeline


my_pipeline = Pipeline()

この Pipeline は、次のように setup.py に登録できます。

Copied!
1
2
3
4
5
6
7
8
9
10
from setuptools import find_packages, setup


setup(
    entry_points={
        'transforms.pipelines': [
            'root = myproject.pipeline:my_pipeline'
        ]
    }
)

上記のコードで、root はエクスポートする Pipeline オブジェクトの名前を指します。myproject.pipeline は Pipeline を含むモジュールを指し、my_pipeline はそのモジュールで定義された Pipeline 属性を指します。