仮想テーブルでの Python トランスフォーム

仮想テーブルを使用すると、データを Foundry に保存せずに、対応するデータプラットフォームのテーブルに対してクエリや書き込みを行えます。transforms-tables ライブラリを使用して、Python トランスフォームで仮想テーブルを操作できます。

Foundry は、仮想テーブルで次の2つのコンピューティングモードをサポートしています。

Foundry ネイティブの Spark コンピューティング： 1つまたは複数のソースの仮想テーブルを入力または出力として使用し、Foundry ネイティブのデータセットの入力および出力と組み合わせることができます。仮想テーブルは、Foundry の Polars や Pandas などの軽量コンピューティングエンジンをサポートしていません。
コンピューティングプッシュダウン： このモードでは、Foundry が軽量コンピューティングを通じて、ソースシステム内の外部コンピューティングをオーケストレーションします。コンピューティングプッシュダウンを使用するには、すべての入力と出力が同じソースに関連付けられた仮想テーブルである必要があり、そのソースがコンピューティングプッシュダウンに対応している必要があります。

コンピューティングプッシュダウンを使用する場合、@incremental デコレーターによるインクリメンタル（差分処理）計算は現在サポートされていません。

前提条件

Python トランスフォームで仮想テーブルを使用するには、次の操作が必要です。

Python リポジトリをアップグレードして最新バージョンにします。
ライブラリタブから transforms-tables をインストールします。

use_external_systems デコレーターを使用するトランスフォームは、仮想テーブルに対応していません。ソースベースの外部トランスフォームに切り替えるか、トランスフォームを、仮想テーブルを使用するものと use_external_systems デコレーターを使用するものに分割してください。

Code Repositories で仮想テーブルを使用すると、その仮想テーブルを入力として使用するトランスフォームは、ソースに設定されたエグレスポリシーに基づいて、ネットワークエグレスを自動的に取得します。

ソースで次の設定を有効にする必要があります。

コードインポート： ソースをコードリポジトリにインポートして使用できるようにします。この設定の詳細については、ソースをコードにインポートするドキュメントを参照してください。
エクスポート制御： 仮想テーブルを使用する Python トランスフォームの入力に、どのセキュリティマーキングおよび組織を許可するかを制御します。詳細については、ソースでエクスポート制御を設定するドキュメントを参照してください。

サイドバーを使用して、出力ソースをリポジトリに忘れずにインポートしてください。

Python トランスフォームで入力および出力として使用するテーブルは、同じソースや同じプラットフォームに属している必要はありません。デフォルトでは、Python トランスフォームは Foundry のコンピューティングを使用してテーブルの読み取りと書き込みを行うため、異なる外部システムを横断したクエリが可能です。コンピューティングを完全に外部システムに委譲する方法の詳細については、コンピューティングプッシュダウンセクションを参照してください。

API リファレンス

TableInput、TableOutput、および利用可能な Table サブクラス（BigQueryTable、DatabricksTable、DeltaTable、FilesTable、IcebergTable、SnowflakeTable など）の詳細については、API 概要のドキュメントを参照してください。

クイックスタートの例

以下は、Snowflake ソースを使用し、仮想テーブルを入力として受け取り、Foundry ネイティブの Spark コンピューティングを使用して仮想テーブルを出力として生成する例です。

Copied!1
2
3
4
5
6
7
8
9
10
11
12
13
14
from transforms.api import transform
from transforms.tables import TableInput, TableOutput, TableTransformInput, TableTransformOutput, SnowflakeTable


@transform.spark.using(
    input_table=TableInput('/path/input_table_name'),
    output_table=TableOutput(
        '/path/output_table_name',  # Foundry の仮想テーブル出力
        'ri.magritte..source.123...',  # 出力ソースの RID
        SnowflakeTable('DATABASE', 'SCHEMA', 'TABLE'),  # 出力の保存場所
    ),
)
def compute(input_table: TableTransformInput, output_table: TableTransformOutput):
    output_table.write_dataframe(input_table.dataframe())

Foundry データセットを仮想テーブルと組み合わせて使用することもできます。以下の例では、Foundry データセットを入力として受け取り、仮想テーブルを出力として外部ソースに書き込みます。

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


@transform.spark.using(
    input_dataset=Input('/path/input_dataset_name'),
    output_table=TableOutput(
        '/path/output_table_name',  # Foundry の仮想テーブル出力
        'ri.magritte..source.123...',  # 出力ソースの RID
        SnowflakeTable('DATABASE', 'SCHEMA', 'TABLE'),  # 出力の保存場所
    ),
)
def compute(input_dataset: TransformInput, output_table: TableTransformOutput):
    output_table.write_dataframe(input_dataset.dataframe())

次の例では、仮想テーブルの入力および出力でコンピューティングプッシュダウンを使用します。

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
from snowflake.snowpark.functions import lit
import snowflake.snowpark as snow
from transforms.api import transform
from transforms.tables import (
    SnowflakeTable,
    TableInput,
    TableOutput,
    SnowflakeInput,
    SnowflakeOutput,
)

@transform.snowflake.using(
    input_table=TableInput('/path/input_table_name'),
    output_table=TableOutput(
        '/path/output_table_name',  # Foundry の仮想テーブル出力
        'ri.magritte..source.123...',  # 出力ソースの RID
        SnowflakeTable('DATABASE', 'SCHEMA', 'TABLE'),  # 出力の保存場所
    ),
)
def compute_in_snowflake(input_table: SnowflakeInput, output_table: SnowflakeOutput):

    # Snowpark DataFrame インスタンスを取得します
    df: snow.DataFrame = input_table.dataframe()

    # データ加工の例 - 新しい列を作成します
    df = df.with_column("NEW_COLUMN", lit("ABC"))

    # 新しいテーブルに書き戻します
    output_table.write(df)

JDBC パーティション分割

JDBC 接続を使用して仮想テーブルのデータを読み取る場合、Python トランスフォームでパーティション分割を設定すると、読み取りを並列化し、テーブルの読み込み性能を向上させることができます。

以下は、input_table の読み込み時にパーティション分割を追加する例です。column_name に基づき、下限1から上限100までの範囲を同じサイズの3個のパーティションに分割します。

Copied!1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
@transform(
    input_table=TableInput("/path/input_table_name"),
    output_df=Output("/path/output_dataset_name")
)
def compute(input_table: TableTransformInput, output_df: TransformOutput):
    input_df = input_table.dataframe(
        options={
            "numPartitions": "3",
            "partitionColumn": "column_name",
            "lowerBound": "1",
            "upperBound": "100"
        }
    )

    output_df.write_dataframe(input_df)

仮想テーブルの入力を読み込むために .dataframe() を呼び出す際は、options パラメーターを通じてパーティション分割の設定を渡す必要があります。partitionColumn は、数値、日付、またはタイムスタンプの列である必要があります。データは、numPartitions で指定された数のパーティションに分散されます。これらのパーティションは、lowerBound オプションと upperBound オプションの間を等間隔に分割した範囲になります。

lowerBound オプションと upperBound オプションは、パーティションの分割幅を計算するためにのみ使用され、行を絞り込むためには使用されません。指定した境界の外側の値を持つ行を含め、テーブルのすべての行が返されます。

利用可能なパーティション分割オプションの詳細については、Spark JDBC ドキュメント ↗を参照してください。

ファイルテンプレート設定ウィザード [ベータ版]

ベータ版

ファイルテンプレート設定ウィザードの仮想テーブル出力は、開発のベータ版段階にあります。開発が進む中で、機能が変更される可能性があります。

仮想テーブルの入力および出力は、Code Repositories のファイルテンプレート設定ウィザードで、仮想テーブルのテンプレート変数タイプを使用して設定できます。仮想テーブル出力を作成する際には、ウィザードの案内に沿って、書き込み先の出力ソースと、仮想テーブルの Foundry 内の場所を選択します。

「出力テーブルのソース」ダイアログで仮想テーブル出力を設定する。

←

PREVIOUSローカル開発のセットアップ

NEXTAPI リファレンス

→