シングルノードコンピューティングの詳細オプション

以下の統合の基本原則は、表形式の Foundry データセットに、pandas DataFrame、Arrow Table、Polars DataFrame、さらには生の Parquet ファイルや CSV ファイルなど、さまざまな形式でアクセスできるということです。これはトランスフォームの概要でも示されています。テーブルをメモリーから Foundry に保存する際は、読み込み時に使用したどの形式でも渡すことができます。

シングルノードコンピューティングエンジン（pandas および Polars）と分散コンピューティング（Spark）のどちらを選ぶかについては、コンピューティングエンジンの選択を参照してください。このページの統合は、特化したユースケース向けにシングルノードの機能を拡張します。

Copied!
1
2
3
import pandas as pd

df = pd.read_parquet(my_input.path())

Apache DataFusion を使用する

データのデシリアライズ手順を省き、元のデータセットファイルをそのままコンピューティングエンジンに渡すほうが簡単な場合があります。my_input.path() を呼び出すと、必要に応じてダウンロードされるファイルのディスク上のパスを取得できます。生のファイルを Foundry に書き戻す際は、次の2つの制限に留意する必要があります。

• この API を通じて Foundry データセットに保存できるのは、Parquet ファイルのみです。
• ファイルは、my_output.path_for_write_table の値が示すフォルダーに配置する必要があります。

両方の条件を満たしたら、次のように、このフォルダーのパスを指定して write_table を呼び出すことができます。

Copied!
1
my_output.write_table(my_output.path_for_write_table)

実際の動作を確認するため、プラットフォームで DataFusion ↗ を使用する方法を示す次の例をご覧ください。

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


@transform.using(my_input=Input('my-input'), my_output=Output('my-output'))
def my_datafusion_transform(my_input, my_output):
    ctx = datafusion.SessionContext()
    table = ctx.read_parquet(my_input.path())
    my_output.write_table(
        table
        .filter(starts_with(col("name"), lit("John")))
        .to_arrow_table()
    )

cuDF を使用する

pandas.DataFrame を使用して統合することもできます。次のスニペットは、軽量トランスフォームで cuDF ↗ を使用する例です。これにより、可能な場合は GPU 上で pandas コードを高度に並列化して実行します。

Copied!
1
2
3
4
5
6
7
8
9
10
11
12
13
14
@transform.using(
    my_input=Input('my-input'),
    my_output=Output('my-output')
).with_resources(
    cpu_cores=4,
    memory_gb=32,
    gpu_type='NVIDIA_T4'
)
def my_cudf_transform(my_input, my_output):
    import cudf  # CUDF は実行時にのみインポートし、CI 中にはインポートしません
    df = cudf.from_pandas(my_input.pandas()[['name']])
    filtered_df = df[df['name'].str.startswith('John')]
    sorted_df = filtered_df.sort_values(by='name')
    my_output.write_table(sorted_df)

上記のスニペットは、Foundry エンロールメントに NVIDIA T4 GPU が搭載され、リソースキューを通じてプロジェクトで利用可能になっていることを前提としています。

cuDF による GPU アクセラレーション対応 Polars

Polars 用の GPU ベースの実行エンジンである cudf-polars ↗ を使用すると、Polars lazy API のクエリを GPU で高速化できます。collect() の呼び出しに pl.GPUEngine() を渡すことで、コードの変更を最小限に抑えながら、既存の Polars lazy API のクエリを GPU 上で実行できます。サポートされる操作と性能特性の詳細については、Polars の GPU サポートに関するドキュメント ↗を参照してください。

GPU アクセラレーション対応 Polars を使用するには、Foundry エンロールメントにサポート対象の NVIDIA GPU が搭載されており、リソースキューを通じてプロジェクトでその GPU を利用できる必要があります。利用可能な GPU の種類については、GPU のプロビジョニングを参照してください。

セットアップ

Polars lazy API で GPU アクセラレーションを使用するには、conda_recipe/meta.yaml ファイルの run 要件に cudf-polars パッケージを追加します。パッケージの追加については、Python ライブラリーのドキュメントを参照してください。

Copied!
1
2
3
4
5
requirements:
  run:
    - python
    - transforms {{ PYTHON_TRANSFORMS_VERSION }}
    - cudf-polars

次に、デコレーターに .with_resources() をチェーンして、トランスフォームで GPU を要求します。

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

@transform.using(
    output=Output("/Users/jsmith/output"),
    my_input=Input("/Users/jsmith/input"),
).with_resources(
    gpu_type="NVIDIA_T4",
    memory_gb=32,
)
def compute(output, my_input):
    ...

Polars GPU エンジンで書き込む

標準の Polars lazy API を使用してクエリを作成し、write_table() に engine="gpu" を渡して、GPU を使用して結果を取得し、書き込みます。

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

@transform.using(
    output=Output("/Users/jsmith/output"),
    my_input=Input("/Users/jsmith/input"),
).with_resources(
    gpu_type="NVIDIA_T4",
    memory_gb=32,
)
def compute(output, my_input):
    lf = my_input.polars(lazy=True)

    result = (
        lf.with_columns(
            (pl.col("a") * pl.col("b")).alias("product"),
            (pl.col("a").pow(2) + pl.col("b").pow(2)).sqrt().alias("norm"),
        )
        .group_by("category")
        .agg(
            pl.col("product").sum().alias("total_product"),
            pl.col("norm").mean().alias("mean_norm"),
            pl.len().alias("n"),
        )
        .sort("total_product", descending=True)
    )

    output.write_table(result, engine="gpu")

GPU エンジンは、Polars の操作の一部のみをサポートしています。完全な一覧については、Polars の GPU サポートに関するドキュメント ↗を参照してください。デフォルトでは、サポートされていないクエリは通知なしで CPU エンジンにフォールバックします。代わりにエラーを発生させるには、raise_on_fail=True を指定して手動で結果を取得し、結果の DataFrame を write_table() に渡します。

Copied!
1
2
df = lf.collect(engine=pl.GPUEngine(raise_on_fail=True))
output.write_table(df)