外部トランスフォーム

外部トランスフォームを使用すると、Python トランスフォームのリポジトリから外部システムに接続できます。

Data Connection ソースは、コードリポジトリに直接インポートできるようになりました。外部システムとやり取りする際には、資格情報ベースのレガシー外部トランスフォームに代わり、この方法を推奨します。

外部トランスフォームは、主に次のいずれかに該当する場合に、バッチ同期、エクスポート、メディア同期のワークフローを実行するために使用されます。

• 既存の Data Connection ソースの種類が利用できない場合。
• 対象となるソースの種類で、必要な機能が利用できない場合。
• Data Connection のユーザーインターフェースで提供される機能に、必要な機能が含まれていない場合。

これらの状況への解決策として、次の方法があります。

• インターネット経由またはプライベートネットワーク内で REST API に接続します。
• データベースに接続し、Data Connection のユーザーインターフェースでは現在実現できないカスタムクエリロジックを構成します。
• 同期またはエクスポート中に、必要に応じてデータを加工します。たとえば、Foundry に書き込む前にファイルをまとめてバッチ化したり、転送中にデータのカスタム暗号化や復号を処理したりします。

仮想テーブルを使用するトランスフォームも、外部トランスフォームと見なされます。これは、仮想化されたデータを格納する外部システムにトランスフォームジョブが接続できる必要があるためです。Python トランスフォームで仮想テーブルを使用するには、以下の手順に沿ってソースを設定してください。

セットアップガイド

このセットアップガイドでは、無料の公開辞書 API ↗ に接続する Python トランスフォームリポジトリを作成する手順を説明します。その後、この API を使った例を通じて、外部トランスフォームのさまざまな機能と、API と組み合わせた使用方法を説明します。

前提条件：Python トランスフォームリポジトリを作成する

このガイドに沿って進める前に、まず Python トランスフォームリポジトリを作成し、チュートリアルで説明されている Python トランスフォームの作成方法を確認してください。Python トランスフォームのすべての機能は、外部トランスフォームでも利用できます。

前提条件：Data Connection ソースを作成する

Python リポジトリから外部システムに接続するには、コードにインポートできる Data Connection ソースを作成する必要があります。このチュートリアルでは、上記の辞書 API に接続する REST API ソースを作成します。

方法1：外部システムのサイドバーでソースを作成する

外部トランスフォームで使用するソースを最も手早く作成するには、Python トランスフォームのコードリポジトリから作成します。リポジトリを初期化したら、次の手順で汎用ソースを設定します。

1. 左側のパネルから、外部システムタブを開きます。

   VS Code ワークスペース内の Python トランスフォームリポジトリで作業している場合、外部システムの設定は、上部ナビゲーションバーの設定タブにあります。

2. 追加 > 新規作成を選択します。

3. ソースの名前と、保存先のプロジェクトを選択します。作成すると、新しいソースが左側のパネルに表示されます。エグレスポリシー、シークレット、エクスポート可能なマーキングは、このパネルから直接設定できます。

4. このチュートリアルでは、辞書 API の api.dictionaryapi.dev に対するエグレスポリシーを追加してください。この API は認証を必要としないため、シークレットは不要です。また、エクスポート制御は現時点では省略できます。ただし、このソースで Foundry のデータ入力を使用するには、エクスポート制御が必要になります。

5. この接続先は REST API のため、組み込みの Python requests クライアントを使用できるように、汎用コネクターを REST API ソースに変換するためのダイアログが自動的に表示されます。

方法2：Data Connection でソースを作成する

Data Connection アプリケーションからソースを作成することも、すでに設定済みのソースを使用することもできます。この方法を使用するには、次の手順に沿って進めます。

1. Foundry 内の Data Connection アプリケーションに移動し、新規ソースを選択します。選択肢の一覧から、REST API を選択します。

2. 概要ページを確認し、右下の続行を選択します。接続ワーカーの選択画面が表示されます。Agent ワーカー接続は外部トランスフォームでは利用できないため、辞書 API に接続するには Foundry ワーカーを選択します。

3. ソースの名前を選び、保存先のプロジェクトを選択します。

4. ドメインセクションに API ソースの接続情報を入力します。辞書 API の例の設定を以下に示します。

5. この例では、必要なエグレスポリシーも作成する必要があります。前の手順を完了すると、ネットワーク接続セクションにポリシーが自動的に提案されます。

6. 保存、続いて保存および続行を選択して、ソースのセットアップを完了します。

ソースを作成しただけでは、リポジトリで利用できるようにはなりません。外部システムタブからこのソースを追加するには、前提条件：ソースをコードにインポートするで説明されているように、コードへのインポートを許可する必要があります。

前提条件：リポジトリに transforms-external-systems ライブラリをインポートする

外部トランスフォームを使用するには、まずリポジトリに transforms-external-systems ライブラリをインポートする必要があります。ライブラリをインストールするには、左側のパネルのライブラリタブで目的のライブラリを検索し、インストールを選択します。

ライブラリのインストールと管理について詳しくはこちら。

前提条件：ソースをコードにインポートする

1. まず、REST API ソースをコードにインポートできるようにする必要があります。この設定を行うには、Data Connection でソースに移動し、接続設定 > コードインポート設定タブに移動します。

2. このソースを Code Repositories にインポートできるようにするオプションをオンにします。このソースをインポートしているコードリポジトリは、このページに表示されます。

3. これでコードリポジトリに戻り、ソースをインポートする準備ができました。リポジトリの左側のパネルに移動し、地球アイコンで示される外部システムタブを選択します。サイドパネル内で追加を選択し、先ほど作成した辞書 API ソースを検索します。このソースを選択し、選択を確定を選択してインポートします。

外部トランスフォームを記述する

Dictionary API ソースをインポートする Python トランスフォームのリポジトリを設定すると、そのソースを使用して外部に接続する Python トランスフォームのコードを記述できるようになります。このページの後半にあるエンドツーエンドの例では、一般的なシステムを対象とした、設定済みの読み取りおよび書き込みワークフローを紹介します。

@external_systems デコレーターをインポートして設定する

外部トランスフォームを使用するには、transforms.external.systems ライブラリから external_systems デコレーターと Source オブジェクトをインポートする必要があります。

Copied!
1
from transforms.external.systems import external_systems, Source

次に、external_systems デコレーターを使用して、トランスフォームに含めるソースを指定します。

Copied!
1
2
3
@external_systems(
    dictionary_api_source=Source("ri.magritte..source.e301d738-b532-431a-8bda-fa211228bba6")
)

ソースは Data Connection で開くためのリンクとして自動的に表示され、リソース識別子の代わりにソース名が表示されます。

ソースの属性と資格情報にアクセスする

ソースをトランスフォームにインポートすると、get_https_connection() メソッドで組み込みの接続オブジェクトを使用して、ソースの属性にアクセスできます。次の例では、前のステップで設定した Dictionary API ソースのベース URL を取得する方法を示します。

Copied!
1
dictionary_api_url = dictionary_api_source.get_https_connection().url

ソースに保存されている追加のシークレットや資格情報にも、ソースからアクセスできます。アクセス可能なシークレット名を確認するには、トランスフォームの左側のパネルに移動します。

コード内でシークレットにアクセスするには、次の構文を使用します。

Copied!
1
dictionary_api_source.get_secret("additionalSecretFoo")

組み込みの HTTP クライアントを使用する

RESTful API を提供するソースでは、ソースオブジェクトを通じて組み込みの HTTPS クライアントを使用できます。このクライアントには、サーバー証明書やクライアント証明書を含め、ソースで指定したすべての詳細が事前に設定されるため、そのまま外部システムへのリクエストを開始できます。

Copied!
1
2
3
4
5
6
dictionary_api_url = dictionary_api_source.get_https_connection().url
dictionary_api_client = dictionary_api_source.get_https_connection().get_client()

# dictionary_api_client は、Python の `requests` ライブラリの設定済み Session オブジェクトです。
# GET リクエストの例：
response = dictionary_api_client.get(dictionary_api_url + "/api/v2/entries/en/" + word, timeout=10)

または、独自のクライアントやソース固有の Python ライブラリを使用し、ソースオブジェクトから属性と資格情報を取得することもできます。ただし、Agent プロキシランタイム（サンセット）を使用してオンプレミスのシステムに接続する場合は、組み込みのクライアントを使用する_必要があります_。このクライアントには、必要な Agent プロキシの設定が自動的に適用されるためです。

例：Dictionary API からデータをインポートする

次の例は、単語のリストを順に処理し、Dictionary API から各単語の発音の文字起こしを取得する、完全なトランスフォームを示しています。

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
from pandas import DataFrame
from transforms.api import transform
from transforms.api import Output, Input, TransformContext, transform_pandas
from transforms.external.systems import external_systems, Source
import pandas as pd
import logging

logger = logging.getLogger(__name__)


@external_systems(
    dictionary_api_source=Source(
        "<source_rid>"
    )
)
@transform_pandas(Output("<output_dataset_rid>"))
def compute(dictionary_api_source) -> DataFrame:
    dictionary_api_url = dictionary_api_source.get_https_connection().url
    dictionary_api_client = dictionary_api_source.get_https_connection().get_client()

    words = ["apple", "dog", "cat"]

    phonetics = []

    for word in words:
        logger.info("Fetching word from api.dictionaryapi.dev : " + word)

        response = dictionary_api_client.get(
            dictionary_api_url + "/api/v2/entries/en/" + word
        ).json()

        phonetics += [{"word": word, "phonetic": response[0]["phonetic"]}]

    return pd.DataFrame(phonetics)

外部トランスフォームで Foundry の入力を使用する

外部トランスフォームでは、Foundry の入力データを使用する必要があることがよくあります。たとえば、テーブル形式のデータセットの各行について、追加のメタデータを収集するために API にクエリを実行したい場合があります。また、Foundry のデータを外部のソフトウェアシステムにエクスポートする必要があるワークフローも考えられます。

このようなケースは、エクスポート制御の対象となるワークフローと見なされます。これは、保護された Foundry のデータが、セキュリティ上の保証が不明で、データ来歴が途切れる別のシステムにエクスポートされる可能性があるためです。ソース接続を設定する際、ソースの所有者は、Foundry からデータをエクスポートしてよいかどうかを指定し、エクスポート可能なセキュリティマーキングと組織を指定する必要があります。Foundry は、開発者がセキュリティ上の意図を明確に設定に反映し、情報セキュリティ責任者が外部システムとやり取りするワークフローの範囲と意図を監査できるように、ガバナンス制御を提供しています。

ソースでエクスポート制御を設定する

エクスポートは、セキュリティマーキングを使用して制御されます。ソースを設定する際、エクスポート設定で、どのセキュリティマーキングと組織のデータを外部システムに安全にエクスポートできるかを指定します。そのためには、Data Connection アプリケーションでソースに移動し、接続設定 > エクスポート設定タブに移動します。次に、このソースへのエクスポートを有効化オプションをオンにして、エクスポートする可能性のあるデータのマーキングと組織を選択します。

この操作には、該当するデータと組織のマーキングを削除する権限が必要です。エクスポートは、Foundry 内のデータからマーキングを削除することと同等と見なされるためです。

次の操作を許可するには、このソースへのエクスポートを有効化の設定をオンにする必要があります。

• このソースをインポートする Python トランスフォームのコードで、データセット、メディアセット、ストリームを入力として使用する。
• このソースに登録された仮想テーブルを Python トランスフォームで使用する。

次に、Dictionary API ソースのエクスポート設定の例を示します。この設定では、追加のセキュリティマーキングが付いていない Palantir 組織のデータを Dictionary API にエクスポートできます。

このシステムに実際にはデータをエクスポートしない場合でも、このソースへのエクスポートを有効化をオンにする必要があります。このシステムへの接続が開かれているコンピューティングジョブに Foundry のデータ入力を許可すると、データがエクスポートされる_可能性がある_ためです。

例：Dictionary API のデータとともに Foundry のインポートを使用する

この例では、ハードコードされた固定のリストの代わりに、単語を含む入力データセットを使用します。また、レスポンスのステータスコードに基づく基本的なエラー処理も示します。

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
43
44
45
46
47
48
49
from pandas import DataFrame
from transforms.api import transform
from transforms.api import Output, Input, TransformContext, transform_pandas
from transforms.external.systems import external_systems, Source, ResolvedSource
import pandas as pd
import logging

logger = logging.getLogger(__name__)


@external_systems(
    dictionary_api_source=Source(
        "<source_rid>"
    )
)
@transform_pandas(
    Output("<output_dataset_rid>"),
    words_df=Input("<input_dataset_rid>"),
)
def compute(dictionary_api_source: ResolvedSource, words_df: DataFrame) -> DataFrame:
    dictionary_api_url = dictionary_api_source.get_https_connection().url
    dictionary_api_client = dictionary_api_source.get_https_connection().get_client()

    words = words_df["word"].tolist()

    phonetics= []

    for word in words:
        logger.info("Fetching word from api.dictionaryapi.dev: " + word)

        response = dictionary_api_client.get(
            dictionary_api_url + "/api/v2/entries/en/" + word
        )

        if response.status_code == 200:
            data = response.json()[0]

            if "phonetic" in data:
                phonetic_transcription = data["phonetic"]
            else:
                logger.warning(f"No phonetic transcription found for {word}.")
                phonetic_transcription = None
        else:
            logger.warning(f"Request for {words} failed with status code {response.status_code}.")
            phonetic_transcription = None

        phonetics += [{"word": word, "phonetic": phonetic_transcription}]

    return pd.DataFrame(phonetics)

エンドツーエンドの例

一般的なシステムを対象とした複雑なエンドツーエンドの例については、次のドキュメントを参照してください。

• REST API ソース
  • OAuth クライアント資格情報グラント
  • 自己署名サーバー証明書
• Foundry API
  • コードから Foundry API を呼び出す
• ファイルベースのシステム
  • Amazon S3
  • SMB
• カスタム Python SDK
  • BigQuery
  • Databricks
  • Snowflake
  • SharePoint
  • SFTP
  • Kafka
• Agent ホストのファイルへのアクセス
  • Directory (SSH)
• JDBC ベースのシステム
  • PostgreSQL
  • Microsoft SQL Server

権限

外部トランスフォームを使用する前に、Data Connection - 権限リファレンスページの内容をよく確認してください。

外部トランスフォームとレガシー外部トランスフォームの比較

外部トランスフォームとレガシー外部トランスフォームのワークフローには、次のような主な違いがあります。

• 外部トランスフォームでは、ソースをインポートするためのタブが常に自動的に表示されます。以前は、情報セキュリティ責任者がリポジトリ設定で外部システムの使用を有効にした後にのみ、エグレスポリシーと資格情報を追加するためのタブが表示されていました。
• 外部接続と入力の使用を許可する設定は、リポジトリ設定にはなくなりました。代わりに、ソースごとに制御されます。
• 資格情報、エグレスポリシー、エクスポート可能なマーキングは、コード内で指定する必要がなくなりました。代わりに、トランスフォームにインポートされたソースからこれらの設定が取得され、ジョブに自動的に適用されます。
  • ソースレベルでこの設定が変更されると、そのソースをインポートするトランスフォームに自動的に反映されます。コードの変更やバージョンの引き上げは必要ありません。これにより、資格情報、エグレス、エクスポート可能なマーキングを一元管理でき、変更は下流のワークフローに即座に伝播します。
  • 変更はビルドの開始時から有効になり、実行中のビルドには影響しません。
• デコレーターが @use_external_systems() から @external_systems() に変更されました。

外部トランスフォームの主な利点は次のとおりです。

• インターネットからアクセスできないシステムへの接続をサポート
• コードの変更を必要とせずに資格情報のローテーションや更新をサポート
• 複数のリポジトリ間での接続設定の共有をサポート
• 一部のソースタイプに対して、そのまま利用できる Python クライアントを提供
• 外部トランスフォームのリポジトリを有効にして管理するためのガバナンスワークフローを改善し、簡素化
• Data Lineage で、外部ソースに接続された外部トランスフォームを可視化

外部トランスフォームに移行する

現在、外部トランスフォームを外部トランスフォームに更新するための自動移行手段はありません。ただし、ほとんどのワークフローでは、必要な手動操作はわずかで済むと見込まれます。

外部トランスフォームに手動で移行するための主な手順は次のとおりです。

1. 既存のレガシー外部トランスフォームのコードで使用されている資格情報、エグレスポリシー、輸出管理マーキングを特定します。
2. 外部トランスフォームから接続したいシステムに接続する Data Connection ソースを特定するか、設定します。これらのソースでコードへのインポートが許可されていることを確認してください。
3. ステップ2の該当するソースを、既存の Python トランスフォームリポジトリにインポートします。
4. ソースへの参照とともに新しい @external_systems() デコレーターをインポートして使用するようにコードを変更し、@use_external_systems() デコレーターのすべての使用箇所を削除します。多くの場合、トランスフォームのロジック内にある資格情報への参照を更新し、新たにインポートするソースから取得した資格情報を参照するようにする必要があります。
5. ブランチで変更をテストし、トランスフォームが引き続き正常にビルドされることを確認します。
6. 更新したトランスフォームのコードをマージした後、リポジトリ設定のトグルをオフにできます。

トランスフォームには、外部トランスフォームとそのレガシーバージョンを混在させることはできません。これを解決するには、すべてのレガシー外部トランスフォームをソースベースの外部トランスフォームに移行するか（推奨）、トランスフォームを複数のトランスフォームに分割します。トランスフォームは、use_external_systems デコレーターを使用するものと、external_systems デコレーターを使用するものに分割できます。

機能

lightweight 外部トランスフォーム

外部トランスフォームは、シングルノード（軽量）のコンピュートエンジンにも対応しています。軽量コンピュートを使用すると、小規模および中規模のデータを処理するトランスフォームの実行速度を大幅に向上させることができます。

次の例は、@external_systems デコレーターとともに @lightweight デコレーターをトランスフォームに追加する方法を示しています。軽量トランスフォームの設定オプションの詳細については、Python トランスフォームのドキュメントを参照してください。

Copied!
1
2
3
4
@lightweight
@external_systems(
    dictionary_api_source=Source("<source_rid>")
)

より詳しい例については、Python でのソースを参照してください。

Agent プロキシポリシーを使用するソース

外部トランスフォームは、Agent プロキシポリシーを使用して、Data Connection の Agent 経由でオンプレミスおよびプライベートにホストされたシステムに接続することをサポートしています。

次の例では、Spark サイドカートランスフォームを使用して PostgreSQL データベースに接続する方法を示します。詳細については、Spark サイドカートランスフォームのドキュメントを参照してください。

次の例よりも単純なユースケースでは、Spark サイドカートランスフォームが不要な場合があります。Spark サイドカートランスフォームを必要としないほかの例は、このページの別の箇所にあります。

Agent ネットワークポリシーを使用してソースを作成する

プライベートネットワークアクセスを使用する外部トランスフォームを設定するには、次の手順を実行します。

1. 直接接続ランタイムを使用して PostgreSQL ソースを作成します： Data Connection ソースを作成する際に、ランタイムの選択肢として直接接続を選択します。
2. ネットワークポリシーを設定します： ソースのネットワーク接続設定で、Data Connection の Agent を経由するルーティングを定義したネットワークポリシーを作成します。Agent プロキシを選択し、プロキシに使用する Agent を選択します。

3. コードのインポートを有効化： コードインポート設定パネルで、このソースを Code Repositories にインポートできるようにするをオンにします。

4. トランスフォーム用のコードリポジトリをセットアップし、ソースをインポートします。

例：Spark サイドカー経由の PostgreSQL 接続

この例は、サイドカーコンテナ内のコンテナ化された Flask API を使用して、PostgreSQL データベースにクエリを実行する方法を示しています。

Dockerfile

Copied!
1
2
3
4
5
6
7
8
FROM python:3.8-slim
WORKDIR /usr/src/app
COPY requirements.txt ./
RUN pip install --no-cache-dir -r requirements.txt
COPY . .
EXPOSE 1234
USER 5001
CMD ["python", "app.py"]

requirements.txt

flask
psycopg2-binary

Flask アプリケーション (app.py)

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
from flask import Flask, jsonify
import psycopg2

app = Flask(__name__)

@app.route('/tables', methods=['POST'])
def tables():
    data = request.get_json()
    password = data.get('password')

    connection_params = {
        "dbname": "<databasename>",
        "user": "<user>",
        "password": password,
        "host": "postgres.com",
        "port": "5432",
    }


    data = {"schema": [], "table": []}
    with psycopg2.connect(**connection_params) as conn:
        with conn.cursor() as cur:
            cur.execute(
                "SELECT table_schema, table_name FROM information_schema.tables WHERE table_type = 'BASE TABLE'"
            )
            tables = cur.fetchall()
            for schema, table in tables:
                data["schema"].append(schema)
                data["table"].append(table)

    return jsonify(data)

トランスフォームの実装

Copied!
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
import requests
from transforms.sidecar import sidecar
from transforms.external.systems import external_systems, Source
from transforms.api import transform_df, Output

@external_systems(
    my_source=Source("<source_rid>")
)
@sidecar(image="simple", tag="0.0.1")
@transform_df(
    Output("<output_dataset_rid>"),
)
def compute(my_source, ctx):
    password = my_source.get_secret("PASSWORD")
    response = requests.post("http://localhost:1234/tables", json={"password": password})
    data = [(response.text,)]
    columns = ["table_schema_json"]
    return ctx.spark_session.createDataFrame(data, columns)

高度なパターン

メモリー使用量を考慮したバッファリングでデータを Parquet ファイルに書き込む

外部トランスフォームが外部ソースから大量のレコードを取得する場合、書き込み前にすべてのレコードをメモリーに保持すると、メモリー不足エラーが発生することがあります。次の BufferedParquetWriter クラスは、設定可能なメモリーしきい値を超えると、蓄積された行を定期的に Parquet ファイルにフラッシュします。このパターンは任意の外部データソースで使用でき、filesystem().open() で生のファイルを書き込み、put_metadata() でスキーマ推論を確定します。

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
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
import gc
import logging
from datetime import datetime

import pyarrow as pa
import pyarrow.parquet as pq
from pympler import asizeof

from transforms.api import Output, transform, LightweightOutput
from transforms.external.systems import external_systems, Source, ResolvedSource

logger = logging.getLogger(__name__)


class BufferedParquetWriter:
    """行をメモリー内のバッファに蓄積し、バッファのサイズが
    設定可能なしきい値を超えると Parquet ファイルに書き出します。"""

    def __init__(self, output: LightweightOutput, schema: pa.Schema, flush_threshold_mb: int = 64):
        self.output = output
        self.schema = schema
        self.flush_threshold_bytes = flush_threshold_mb * 1024 * 1024
        self.buffer = []
        self.buffer_size_bytes = 0
        self.file_index = 0
        self.row_size_estimate = None

    def append(self, row: dict):
        """バッファに行を追加します。バッファがしきい値を超えた場合は、ディスクに書き出します。"""
        self.buffer.append(row)

        # 最初の行から行のサイズを推定し、以降はその推定値を使用します
        if self.row_size_estimate is None:
            self.row_size_estimate = asizeof.asizeof(row)
        self.buffer_size_bytes += self.row_size_estimate

        if self.buffer_size_bytes >= self.flush_threshold_bytes:
            self.flush()

    def flush(self):
        """バッファ内の行を Parquet ファイルに書き込み、バッファをクリアします。"""
        if not self.buffer:
            return
        self.file_index += 1
        table = pa.Table.from_pylist(self.buffer, schema=self.schema)
        timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
        filename = f"output_{timestamp}_{self.file_index}.parquet"
        with self.output.filesystem().open(filename, "wb") as f:
            pq.write_table(table, f)
        logger.info(f"Flushed {len(self.buffer)} rows to {filename}")
        self.buffer = []
        self.buffer_size_bytes = 0
        gc.collect()

    def finalize(self):
        """残りの行を書き出し、スキーマ推論を開始します。"""
        self.flush()
        if self.file_index > 0:
            self.output.put_metadata()


# 出力レコードのスキーマを定義します
OUTPUT_SCHEMA = pa.schema([
    ("id", pa.string()),
    ("name", pa.string()),
    ("value", pa.float64()),
])


@external_systems(
    my_source=Source("<source_rid>")
)
@transform.using(
    output=Output("<output_dataset_rid>"),
).with_resources(
    memory_gb=4,
)
def compute(my_source: ResolvedSource, output: LightweightOutput):
    output.set_mode("replace")
    writer = BufferedParquetWriter(output, OUTPUT_SCHEMA, flush_threshold_mb=64)

    # 以下のループをデータ取得ロジックに置き換えます。
    # たとえば、API のページ、データベースカーソルの行、
    # またはメッセージキューのメッセージを順に処理します。
    # append() を呼び出すたびにライターにデータが追加され、
    # バッファがしきい値を超えると、自動的に Parquet ファイルに書き出されます。
    for record in fetch_records_from_source(my_source):
        writer.append({
            "id": record["id"],
            "name": record["name"],
            "value": record["value"],
        })

    # バッファ内に残っている行を書き込み、スキーマ推論を完了します。
    writer.finalize()

データセットを CSV としてエクスポートする

たとえば CSV 形式を要求する外部システムにファイルを提供するために、Foundry からデータを CSV としてエクスポートする必要がある場合は、入力データセットを読み取り、filesystem().open() を使用して出力に CSV ファイルを書き込む外部トランスフォームを使用できます。その後、出力データセットを Data Connection 経由でエクスポートできます。

Copied!
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
import csv

from transforms.api import Input, Output, transform, LightweightInput, LightweightOutput


@transform.using(
    source_data=Input("<input_dataset_rid>"),
    csv_output=Output("<output_dataset_rid>"),
)
def compute(source_data: LightweightInput, csv_output: LightweightOutput):
    OUTPUT_FILENAME = "export.csv"

    # 入力データセットを Polars DataFrame として読み込みます。
    df = source_data.polars()
    fieldnames = df.columns

    # データを CSV ファイルとして書き込みます。
    with csv_output.filesystem().open(OUTPUT_FILENAME, "w") as f:
        writer = csv.DictWriter(f, fieldnames=fieldnames)
        writer.writeheader()
        for row in df.iter_rows(named=True):
            writer.writerow(row)