外部トランスフォーム [レガシー]

Code Repositories の Python トランスフォームを使用すると、コードから外部システムに直接接続できます。たとえば、外部システムとのインタラクションに複雑なロジックが必要な場合や、サードパーティーが提供するソフトウェア開発キット (SDK) を使用する場合に役立ちます。REST API を使用したスケジュール済みの同期やエクスポートには、外部トランスフォームの使用を推奨します。

ソースベースの外部トランスフォームは、コードで外部システムに接続するためのサポート対象の方法です。このページで説明するレガシー外部トランスフォームと比べて、インターネットからアクセスできないシステムへの接続、コードを変更せずに行う資格情報のローテーション、仮想テーブルの使用といった利点があります。そのほかの利点や、2つの方式のワークフローの違いについては、外部トランスフォームとレガシー外部トランスフォームの比較を確認してください。

続行する前に、必ず Python トランスフォームのコードリポジトリを作成し、以下の手順に従って設定してください。

Code Repositories による外部接続の設定を許可する

このセクションでは、外部トランスフォームを使用するために必要な設定を説明します。

外部システムとのインタラクションを有効にする

作成したコードリポジトリで、情報セキュリティ責任者ロールを持つ Foundry ユーザーがリポジトリの設定タブに移動し、リポジトリ > 外部システムに移動して、このリポジトリから外部システムへのアクセスを許可するオプションをオンにします。

情報セキュリティ責任者は Foundry のデフォルトロールです。Control Panel のエンロールメント権限で、ユーザーに情報セキュリティ責任者ロールを付与できます。

データセットの入力を有効にする

情報セキュリティ責任者ロールを持つ Foundry ユーザーは、任意で外部トランスフォームにおける入力の使用を許可できます。

外部システムと通信できるトランスフォームの入力としてデータセットを使用すると、そのデータセット内のあらゆるデータが Foundry の外部に送信される可能性があります。情報セキュリティ責任者ロールを持つ Foundry ユーザーが、このリポジトリの外部トランスフォームへの入力として許可するデータのセキュリティマーキングと組織マーキングを選択します。

たとえば、Palantir 組織内に機密マーキングがついたデータセットがあるとします。このデータセットを外部トランスフォームで使用するには、情報セキュリティ責任者が、コードリポジトリ設定のステップ3、外部システムでの Foundry 入力の使用を設定するで、機密マーキングと Palantir 組織の両方を追加する必要があります。

transforms-external-systems ライブラリを追加する

コードリポジトリで外部システムとのインタラクションを有効にしたら、コードインターフェースの左側にあるライブラリサイドバータブから transforms-external-systems ライブラリを追加する必要があります。

VS Code ワークスペースで Python トランスフォームリポジトリを使用している場合は、上部のナビゲーションバーの設定タブに移動し、外部システムを選択すると、出口と資格情報の設定にアクセスできます。

エグレスと資格情報を設定する

ライブラリを追加すると、コードインターフェースの左側にある新しい egressと認証情報タブにアクセスできます。

このタブでは、コードから呼び出すエンドポイントに必要なネットワークエグレスポリシーを設定し、リポジトリで使用する資格情報を追加できます。すべてのエンドポイントで資格情報が必要なわけではありません。

ネットワークエグレスポリシーと資格情報を使用する

エグレスポリシーを追加する

Foundry では、ユーザーが作成したコードから開始されるすべての接続について、外部の宛先に接続する際にネットワークエグレスポリシーに紐づける必要があります。ネットワークエグレスポリシーは、ユーザーが作成したコードから開始される、許可された接続が Foundry の外部の宛先に到達できるようにします。

エンロールメントでセルフサービスエグレスが有効になっている場合は、Control Panel でエグレスポリシーを作成および管理できます。接続に新しいネットワークルートの追加が必要な場合は、エグレスポリシーの詳細を添えて、情報セキュリティ責任者または Palantir サポートにお問い合わせください。お使いの Foundry インスタンスでセルフサービスエグレスが有効になっていない場合は、Palantir サポートに連絡して、内部でのエンドポイントの有効化とポリシーの設定を依頼してください。

外部トランスフォームでエグレスポリシーを使用する前に（グローバルか非グローバルかを問わず）、コードリポジトリの egressと認証情報パネルにある出口タブからポリシーをインポートする必要があります。出口タブには、権限を持つすべてのポリシーが表示され、次の3つのカテゴリーに分類されます。

• インポート済み： 現在のコードリポジトリ内のトランスフォームから使用できるネットワークエグレスポリシーです。
• インポート可能： ネットワークエグレスポリシーへのプロジェクト参照の追加が必要です。
• アクセス要求が必要： ポリシーに対する閲覧者権限はありますが、インポーター権限はありません。

このタブで、目的のネットワークエグレスポリシーを検索します。ポリシーがインポート可能と指定されている場合は、そのポリシーを選択し、egressポリシーをプロジェクトにインポートを選択します。

目的のネットワークエグレスポリシーがアクセス要求が必要と指定されている場合は、情報セキュリティ責任者ロールを持つ Foundry ユーザーが、Control Panel のネットワークのエグレスページで、対象のユーザーにインポーターロールを付与します。インポーターロールを付与するには、情報セキュリティ責任者がまず目的のネットワークポリシーを選択し、次にアクション > 共有の管理を選択して、対象のユーザーをポリシーのインポーターとして追加します。

資格情報を追加する

外部システムへのアクセスに、認可された資格情報（ユーザー名とパスワード）が必要な場合は、egressと認証情報サイドパネルの資格情報タブで資格情報を追加し、コードで使用できます。

証明書を渡す

外部システムに SSL/TLS 証明書を渡す必要がある場合は、証明書を資格情報に保存し、その資格情報を外部トランスフォームのロジックで使用できます。

外部トランスフォームのロジックを記述する

これで、外部システムにアクセスする Python データトランスフォームを記述する準備ができました。外部トランスフォームを記述する前に、Python データトランスフォームを記述するための基本的な手順を確認することを推奨します。

transforms.external.systems パッケージから、必須の use_external_systems デコレーターと、EgressPolicy および Credential（必要な場合）の入力をインポートします。transforms-api から、トランスフォームデコレーターと Output をインポートします。外部ソースで資格情報が必要ない場合、ロジックに Credential の入力を追加する必要はありません。

次の例は、use_external_systems デコレーターの使用方法を示しています。

Copied!
1
2
3
4
5
6
7
8
9
10
11
12
from transforms.api import transform, Output
from transforms.external.systems import use_external_systems, EgressPolicy, Credential

@use_external_systems(
    egress=EgressPolicy('<policy RID>'),
    creds=Credential('<credential RID>')
)
@transform(
    output=Output('/path/to/output/dataset')
)
def compute(egress, creds, ...):
    # ...

次に、API にアクセスするシンプルなトランスフォームを設定できます。

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


@use_external_systems(
    egress=EgressPolicy('<policy RID>'),
    creds=Credential('<credential RID>')
)
@transform(
    output=Output('/path/to/output/dataset')
)
def compute(egress, output, creds):
    username = creds.get('username')
    password = creds.get('password')
    response = requests.get('https://<API URL>', auth=(username, password), timeout=10).text
    with output.filesystem().open('response.json', 'w') as f:
        f.write(response)

以下の例に示すように、1つのトランスフォームで複数のエグレスポリシーや資格情報を指定することもできます。

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

@use_external_systems(
    egress1=EgressPolicy('<first policy RID>'),
    egress2=EgressPolicy('<second policy RID>'),
    creds1=Credential('<first credential RID>'),
    creds2=Credential('<second credential RID>')
)
@transform(
    output=Output('/path/to/output/dataset')
)
def compute(egress1, egress2, creds1, creds2, ...):
    # ...

Foundry データセットの入力を追加する

場合によっては、Foundry の入力データを処理する外部トランスフォームを記述すると便利です。たとえば、API にクエリを実行して、表形式のデータセットの各行について追加のメタデータを収集できます。また、Foundry のデータを外部ソフトウェアシステムにミラーリングする必要があるワークフローも考えられます。

このようなケースは、セキュリティが保証されるか不明で、データ来歴も途切れる別のシステムに、安全に保護された Foundry のデータをエクスポートする可能性が生じるため、エクスポート制御の対象となるワークフローとみなされます。エクスポートワークフローのセキュリティを評価し、Foundry から外部に出るデータが外部システムで適切に保護されるようにすることは、トランスフォーム開発者の責任です。Foundry は、開発者がセキュリティ上の意図を明確に設定に反映し、情報セキュリティ責任者が外部システムとやり取りするワークフローの範囲と意図を監査できるように、ガバナンス制御を提供しています。

エクスポート制御を有効にするには、外部トランスフォームに ExportControl を適用します。ExportControl は、エクスポート対象のセキュリティマーキングと組織IDのリストを受け取ります。これにより、上流データに追加のセキュリティマーキングがついている場合、トランスフォームジョブは必ず失敗します。コードリポジトリでサポートされるエクスポート可能なマーキングを増やすには、情報セキュリティ責任者がリポジトリの設定を調整する必要があります。設定 > リポジトリタブに移動します。次に、外部システムセクションのステップ3で、このリポジトリで外部システムでの Foundry 入力の使用を許可するを選択します。

たとえば、エクスポートを行う外部トランスフォームを使用して、Foundry データセットで指定された画像 URL から画像をダウンロードできます。

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
from transforms.api import transform, Input, Output
from transforms.external.systems import use_external_systems, ExportControl, EgressPolicy
from pyspark.sql.functions import udf
import shutil

@use_external_systems(
    export_control=ExportControl(markings=['<marking ID>']),
    egress=EgressPolicy(<policy RID>),
)
@transform(
    images_output=Output('/path/to/output/dataset'),
    image_urls=Input('/path/to/input/dataset'),
)
def compute(export_control, egress, images_output, image_urls):

    @udf
    def download(name, url):
        response = requests.get(url, stream=True)
        with images_output.filesystem().open(f'{name}.jpg', mode='wb') as out_file:
            shutil.copyfileobj(response.raw, out_file)

        return True

    image_urls.dataframe().withColumn('downloaded', download(col('Name'), col('Url'))).collect()

Foundry メディアセットの出力を追加する

外部トランスフォームを使用して画像、音声、PDF などのメディアファイルを取り込む場合、ファイルをメディアセットに書き出すことができます。既存のメディアセットにメディアアイテムを書き込むには、指定したメディアセットの出力にプログラムでアイテムを配置します。

Copied!
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
from transforms.api import transform
from transforms.mediasets import MediaSetOutput
from transforms.external.systems import use_external_systems
import requests
import tempfile


@use_external_systems()
@transform(
    media_set_output=MediaSetOutput('/path/to/output/media/set')
)
def compute(media_set_output):
    response = requests.get('https://<API URL>')
    fname = 'my_image.png'

    with tempfile.NamedTemporaryFile() as tmp:
        tmp.write(response.content)
        tmp.flush()

        with open(tmp.name, 'rb') as tmp_read:
            media_set_output.put_media_item(tmp_read, path=fname)

外部システムに接続するその他の方法

API サービスとやり取りする

外部システムに API サービスが存在する場合、プログラムでそのサービスにアクセスして使用できます。たとえば、外部トランスフォームから Mapbox Static Images 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
from transforms.external.systems import EgressPolicy, use_external_systems, Credential
from transforms.api import transform, Output
from mapbox import Static


@use_external_systems(
    egress=EgressPolicy('<policy RID>'),
    mapbox_creds=Credential('<credential RID>')
)
@transform(
    output=Output('/Users/username/datasets/example_mapbox'),
)
def compute(output, egress, mapbox_creds):
    mapbox_access_token = mapbox_creds.get('mapbox-token')
    client = Static(access_token=mapbox_access_token)
    london = client.image('mapbox.satellite', lat=51.5072, lon=0.0, z=12)
    with output.filesystem().open('london.log', 'w') as f:
        f.write(f'response code: {london.status_code}')
    with open('london.png', 'wb') as f:
        f.flush()
    with output.filesystem().open('london.png', 'wb') as f:
        f.write(london.content)

データセットのファイルとやり取りする

通常のトランスフォームと同様に、プログラムでファイルを開いたり、ストリーミングしたり、操作したりできます。非構造化ファイルの読み取りと書き込みについては、詳しくはこちらをご覧ください。

インクリメンタル（差分処理）

通常のトランスフォームと同様に、外部トランスフォームをインクリメンタルに実行できます。

インクリメンタルな外部トランスフォームでは、実行間で状態を維持する必要があることがよくあります。通常のインクリメンタルトランスフォームでは、入力ファイルが追跡され、処理済みの最新のトランザクションと比較されます。インクリメンタルな外部トランスフォームでは、インクリメンタル処理の状態を別の方法で明示的に保存する必要があります。1つの方法は、以下の例に示すように、インクリメンタルトランスフォームの最新の状態を保持する状態ファイルを保存することです。

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
from transforms.api import transform, Input, Output, configure, incremental
from pyspark.sql import Row
from pyspark.sql import functions as F
from pyspark.sql import types as T
import logging
import time
import json
log = logging.getLogger(__name__)

@incremental()
@configure(profile=["KUBERNETES_NO_EXECUTORS"])
@use_external_systems(
    egress=EgressPolicy('<policy RID>'),
    creds=Credential('<credential RID>')
)
@transform(
    out=Output("<output_dataset>"),
)
def compute(ctx, out, egress, creds):
    # ファイルを読み書きするために、出力データセットのファイルシステムを取得します。
    out_fs = out.filesystem()
    state_filename = "_state.json" # 先頭に "_" を付けると、隠しファイルとして扱われ、データセットのプレビューに表示されなくなります。
    state = {"last_seen" : 0} # 任意の初期状態

    # 出力データセットから状態の取得を試みます。
    try:
        with out_fs.open(state_filename, mode='r') as state_file:
            data = json.load(state_file)
            # 取得した状態を検証します。
            state = data
            logging.info(f"state file found, continuing from : {data}")

    except Exception as e:
        logging.warn("state file not found, starting over from default state")


    # ここに、API 呼び出し、カスタム処理、データフレームの生成と保存、または出力データセットへのファイルの直接保存を行うロジックを記述します。
    # 例として、以下ではデータフレームを生成し、出力データセットに保存します（API 呼び出しでデータを取得した場合を想定しています）。
    out.write_dataframe(get_dataframe(ctx))
    # 次の反復に向けて状態を更新します。
    state["last_seen"] = 1 + state["last_seen"]

    # 新しい状態を出力データセットに保存します。
    with out_fs.open(state_filename, "w") as state_file:
        json.dump(state, state_file)


# この関数は例示用のデータフレームを生成します。この手順は本番環境の実装には該当しません。
def get_dataframe(ctx):
    # データフレームのスキーマを定義します。
    schema = T.StructType([
        T.StructField("name", T.StringType(), True),
        T.StructField("age", T.IntegerType(), True),
        T.StructField("city", T.StringType(), True)
    ])

    # 行のリストを作成します。
    data = [("Alice", 25, "New York"),
            ("Bob", 30, "San Francisco"),
            ("Charlie", 35, "London")]

    # PySpark データフレームを作成します。
    df = ctx.spark_session.createDataFrame(data, schema=schema)