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

はじめに

はじめての場合は、まずバッチパイプラインのチュートリアルを検討してください

以下の手順では、Python によるシンプルなデータ変換を説明します。データ変換をはじめたばかりの場合は、まず Pipeline Builder または Code Repositories のバッチパイプラインのチュートリアルに取り組むことを検討してください。

このチュートリアルでは、Python トランスフォームを使って、最近発見された隕石のスプレッドシートを、分析に使えるデータセットに変換する方法を説明します。

データセットについて

このチュートリアルでは、NASA のオープンデータポータル ↗ のデータを使用します。次のサンプルデータセットを使って、ご自身のコードリポジトリでチュートリアルに沿って進めることができます。

Download meteorite_landings

このデータセットには、地球上で発見された隕石に関するデータが含まれており、扱いやすいようにクレンジングされています。

データセットには、隕石ごとの名前、質量、機密情報区分、そのほかの識別情報に加えて、発見された年と場所の座標が含まれています。Foundry にアップロードする前に CSV を開いてデータを確認することをお勧めします。

Python コードリポジトリを設定する

Python コードリポジトリを作成してはじめます。

プロジェクトに移動し、+ 新規 > コードリポジトリを選択します。
Repositoryタイプセクションで、パイプラインを選択します。
言語として Python を選択します。
リポジトリを初期化するを選択します。

任意：ローカルの Python リポジトリを使用する

または、次の手順でローカルの Python リポジトリを Code Repositories にコピーできます。

上記の説明に従って、新しい Python コードリポジトリを作成します。
ローカルリポジトリで、以前の Git の origin を削除します（たとえば、GitHub からクローンした場合）: git remote remove origin
コードリポジトリの Git リモート URL を追加します: git remote add origin <repository_url>

コードリポジトリの URL は、GitHub インターフェースの右上にあります。緑色のクローンするボタンを選択し、Git リモート URL をコピーします。git remote -v を実行してコードリポジトリの URL を取得し、確認します。
Code Repositories の現在の master ブランチ（または任意の別のブランチ）をローカルブランチにマージします: git merge master

refusing to merge unrelated histories に関するエラーが発生した場合は、次のコマンドを実行します: git merge master --allow-unrelated-histories。これにより、以前のリモート GitHub リポジトリに関連付けられた現在の Git 履歴が削除されます。

このマージにより、Code Repositories でコミットや変更を行うために必要なファイルがローカルリポジトリに追加されます。
新しいブランチを作成して名前を付けます（たとえば testbranch）: git checkout -b testbranch。
変更を加え、このブランチにコミットします。
git push を実行し、新しいブランチが Code Repositories のインターフェースに表示されることを確認します。チェックが成功していることを確認します。

Code Repositories でのローカル開発については、詳しくはこちら。

Python のデータ変換を記述する

Python トランスフォームのリポジトリに移動します。デフォルトの examples.py ファイルには、はじめる際に役立つサンプルコードが含まれています。コンピュートエンジンは pandas、Polars、Spark から選択できます。コンピュートエンジンの選択については、コンピュートエンジンの比較のドキュメントを参照してください。まず、分析を整理するために、src/myproject/datasets に新しいファイルを作成し、meteor_analysis.py という名前を付けます。必要な関数とクラスを必ずインポートしてください。meteor_landings データセットを入力として受け取り、出力として meteor_landings_cleaned を作成するデータ変換を定義します。

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


@transform.using(
    # 出力データセットのパスに置き換えます
    output=Output("/Users/jsmith/meteorite_landings_cleaned"),
    # 入力データセットのパスに置き換えます
    meteorite_landings=Input("/Users/jsmith/meteorite_landings"),
)
def clean(output, meteorite_landings):
    df = meteorite_landings.polars()
    # データ変換ロジック
    output.write_table(df)

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


@transform.using(
    # 出力データセットのパスに置き換えます
    output=Output("/Users/jsmith/meteorite_landings_cleaned"),
    # 入力データセットのパスに置き換えます
    meteorite_landings=Input("/Users/jsmith/meteorite_landings"),
)
def clean(ctx, output, meteorite_landings):
    conn = ctx.duckdb().conn
    # SQL を使用したデータ変換ロジック
    query = conn.sql("SELECT * FROM meteorite_landings")
    output.write_table(query)

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


@transform.using(
    # 出力データセットのパスに置き換えます
    output=Output("/Users/jsmith/meteorite_landings_cleaned"),
    # 入力データセットのパスに置き換えます
    meteorite_landings=Input("/Users/jsmith/meteorite_landings"),
)
def clean(output, meteorite_landings):
    df = meteorite_landings.pandas()
    # データ変換ロジック
    output.write_table(df)

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


@transform.spark.using(
    # 出力データセットのパスに置き換えます
    output=Output("/Users/jsmith/meteorite_landings_cleaned"),
    # 入力データセットのパスに置き換えます
    meteorite_landings=Input("/Users/jsmith/meteorite_landings"),
)
def clean(output, meteorite_landings):
    df = meteorite_landings.dataframe()
    # データ変換ロジック
    output.write_dataframe(df)

ここで、入力データセットを、1950年より後の「Valid」の隕石に絞り込みたいとします。データ変換のロジックを更新し、nametype と year で隕石を絞り込みます。

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


@transform.using(
    output=Output("/Users/jsmith/meteorite_landings_cleaned"),
    meteorite_landings=Input("/Users/jsmith/meteorite_landings"),
)
def clean(output, meteorite_landings):
    df = (
        meteorite_landings.polars()
        .filter(pl.col("nametype") == "Valid")
        .filter(pl.col("year") >= 1950)
    )
    output.write_table(df)

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

@transform.using(
    output=Output("/Users/jsmith/meteorite_landings_cleaned"),
    meteorite_landings=Input("/Users/jsmith/meteorite_landings"),
)
def clean(ctx, output, meteorite_landings):
    conn = ctx.duckdb().conn
    query = conn.sql("""
        SELECT *
        FROM meteorite_landings
        WHERE nametype = 'Valid' AND year >= 1950
    """)
    output.write_table(query)

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


@transform.using(
    output=Output("/Users/jsmith/meteorite_landings_cleaned"),
    meteorite_landings=Input("/Users/jsmith/meteorite_landings"),
)
def clean(output, meteorite_landings):
    df = meteorite_landings.pandas()
    df = df[
        (df["nametype"] == "Valid") &
        (df["year"] >= 1950)
    ]
    output.write_table(df)

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


@transform.spark.using(
    output=Output("/Users/jsmith/meteorite_landings_cleaned"),
    meteorite_landings=Input("/Users/jsmith/meteorite_landings"),
)
def clean(output, meteorite_landings):
    df = (
        meteorite_landings.dataframe().filter(
            meteorite_landings.dataframe().nametype == 'Valid'
        ).filter(
            meteorite_landings.dataframe().year >= 1950
        )
    )
    output.write_dataframe(df)

出力データセットをビルドする

結果のデータセットをビルドするには、変更をコミットし、右上の Build を選択します。Code Repositories でのデータセットのビルドについては、シンプルなバッチパイプラインを作成するチュートリアルを参照してください。

データ変換を追加する

Python トランスフォームでは、1つの Python ファイルで複数の出力データセットを作成できます。

さらに、その種類の中で特に大きな隕石だけに絞り込みたいとします。そのためには、次の操作が必要です。

隕石の種類ごとの平均質量を求めます。
隕石ごとの質量を、その隕石の種類の平均質量と比較します。

まず、隕石の種類ごとの平均質量を求めるデータ変換を meteor_analysis.py に追加します。このデータ変換は、meteor_landings データセットを入力として受け取り、出力として meteorite_stats を作成します。

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


@transform.using(
    # 出力データセット名は一意である必要があります
    output=Output("/Users/jsmith/meteorite_stats"),
    meteorite_landings=Input("/Users/jsmith/meteorite_landings"),
)
def stats(output, meteorite_landings):
    df = (
        meteorite_landings.polars()
        .groupby("class")
        .agg(pl.col("mass").mean().alias("avg_mass_per_class"))
    )
    output.write_table(df)

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


@transform.using(
    # 出力データセット名は一意である必要があります
    output=Output("/Users/jsmith/meteorite_stats"),
    meteorite_landings=Input("/Users/jsmith/meteorite_landings"),
)
def stats(output, meteorite_landings):
    df = (
        meteorite_landings.pandas()
            .groupby("class", as_index=False)["mass"]
            .mean()
            .rename(columns={"mass": "avg_mass_per_class"})
    )
    output.write_table(df)

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


@transform.spark.using(
    # 出力データセット名は一意である必要があります
    output=Output("/Users/jsmith/meteorite_stats"),
    meteorite_landings=Input("/Users/jsmith/meteorite_landings"),
)
def stats(output, meteorite_landings):
    df = (
        meteorite_landings.dataframe().groupBy("class").agg(
            F.mean("mass").alias("avg_mass_per_class")
        )
    )
    output.write_dataframe(df)

次に、隕石ごとの質量を、その隕石の種類の平均質量と比較するデータ変換を作成します。このデータ変換に必要な情報は、このチュートリアルで作成した meteorite_landings テーブルと meteorite_stats テーブルに分散しています。以下に示すように、2個のデータセットを結合し、結果のデータセットを絞り込んで、平均を上回る質量の隕石を見つける必要があります。

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


@transform.using(
    output=Output("/Users/jsmith/meteorite_enriched"),
    meteorite_landings=Input("/Users/jsmith/meteorite_landings"),
    meteorite_stats=Input("/Users/jsmith/meteorite_stats"),
)
def enriched(output, meteorite_landings, meteorite_stats):
    df_landings = meteorite_landings.polars()
    df_stats = meteorite_stats.polars()

    enriched = df_landings.join(df_stats, on="class", how="inner")
    enriched = enriched.with_columns(
        (pl.col("mass") > pl.col("avg_mass_per_class")).alias("greater_mass")
    )
    result = enriched.filter(pl.col("greater_mass"))
    output.write_table(result)

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


@transform.spark.using(
    output=Output("/Users/jsmith/meteorite_enriched"),
    meteorite_landings=Input("/Users/jsmith/meteorite_landings"),
    meteorite_stats=Input("/Users/jsmith/meteorite_stats"),
)
def enriched(output, meteorite_landings, meteorite_stats):
    df_landings = meteorite_landings.pandas()
    df_stats = meteorite_stats.pandas()

    enriched = df_landings.merge(df_stats, on="class", how="inner")
    enriched["greater_mass"] = enriched["mass"] > enriched["avg_mass_per_class"]
    result = enriched[enriched["greater_mass"]]
    output.write_table(result)

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


@transform.spark.using(
    output=Output("/Users/jsmith/meteorite_enriched"),
    meteorite_landings=Input("/Users/jsmith/meteorite_landings"),
    meteorite_stats=Input("/Users/jsmith/meteorite_stats"),
)
def enriched(output, meteorite_landings, meteorite_stats):
    df_landings = meteorite_landings.dataframe()
    df_stats = meteorite_stats.dataframe()

    enriched_together = df_landings.join(df_stats, "class")
    greater_mass=enriched_together.withColumn(
        'greater_mass', (enriched_together.mass > enriched_together.avg_mass_per_class)
    )
    result = greater_mass.filter("greater_mass")
    output.write_dataframe(result)

これで、結果の meteorite_enriched データセットを Contour で探索して、さらに分析できます。

データ変換を複数の入力に適用する

ここまでで、すべての種類の隕石のうち、平均を上回る質量の隕石を含むデータセットを作成しました。次のステップでは、隕石の種類ごとに個別のデータセットを作成します。Python トランスフォームでは、for ループを使って、同じデータ変換を隕石の種類ごとに適用できます。同じデータ変換を異なる入力に適用する方法については、トランスフォームの生成のセクションを参照してください。

src/myproject/datasets に新しいファイルを作成し、meteor_class.py という名前を付けます。meteor_analysis.py ファイルに引き続きデータ変換のコードを記述することもできますが、このチュートリアルでは、データ変換のロジックを分けるために新しいファイルを使用します。

隕石の種類ごとに個別のデータセットを作成するには、meteorite_enriched データセットをクラスで絞り込みます。次に、分析したい隕石の種類ごとに同じデータ変換のロジックを適用する transform_generator 関数を定義します。

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


def transform_generator(sources):
    transforms = []
    for source in sources:
        @transform.using(
            output=Output(f'/Users/jsmith/meteorite_{source}'),
            my_input=Input('/Users/jsmith/meteorite_enriched')
        )
        def filter_by_source(output, my_input, source=source):
            df = my_input.polars()
            result = df.filter(pl.col("class") == source)
            output.write_table(result)
        transforms.append(filter_by_source)
    return transforms

TRANSFORMS = transform_generator(["L6", "H5", "H4"])

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


def transform_generator(sources):
    transforms = []
    for source in sources:
        @transform.using(
            output=Output(f'/Users/jsmith/meteorite_{source}'),
            my_input=Input('/Users/jsmith/meteorite_enriched')
        )
        def filter_by_source(output, my_input, source=source):
            df = my_input.pandas()
            result = df[df["class"] == source]
            output.write_table(result)
        transforms.append(filter_by_source)
    return transforms

TRANSFORMS = transform_generator(["L6", "H5", "H4"])

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


def transform_generator(sources):
    transforms = []
    for source in sources:
        @transform.spark.using(
            output=Output(f'/Users/jsmith/meteorite_{source}'),
            my_input=Input('/Users/jsmith/meteorite_enriched')
        )
        def filter_by_source(output, my_input, source=source):
            result = my_input.dataframe().filter(my_input["class"] == source)
            output.write_table(result)
        transforms.append(filter_by_source)
    return transforms

TRANSFORMS = transform_generator(["L6", "H5", "H4"])

これにより、隕石データセットをクラスで絞り込むデータ変換が作成されます。関数のスコープ内で source パラメーターを捕捉するために、filter_by_source 関数に source=source を渡す必要があります。

生成したトランスフォームはリスト変数に代入する必要があります

meteor_analysis.py ファイルに作成した最初のデータ変換では、プロジェクトのパイプラインにトランスフォームを追加するための設定は不要でした。これは、デフォルトの Python プロジェクト構造では、自動登録を使って datasets フォルダー内のすべてのトランスフォームオブジェクトを見つけるためです。

自動登録を使って、この最後のデータ変換をプロジェクトのパイプラインに追加するには、生成したトランスフォームをリストとして変数に追加する必要があります。上記の例では、変数 TRANSFORMS を使用しました。自動登録とトランスフォームジェネレーターについては、Python トランスフォームのドキュメントのトランスフォームの生成のセクションを参照してください。

←

PREVIOUS概要

NEXTPython バージョンのサポート

→