エクスポート

Data Connection は、Foundry から外部システムへのデータセットおよびストリームのエクスポートをサポートしています。これは、さまざまな用途で役立ちます。

• データパイプラインを使用して Foundry でクレンジングおよび加工したデータを、データウェアハウスやデータレイクなどのシステムに同期できます。このパターンについては、「Palantir Foundry がデータプラットフォームを支える仕組み」↗で詳しく説明しています。
• バッチデプロイメントを使用して作成した機械学習モデルの推論結果をほかのシステムにエクスポートし、組織全体で ML プロジェクトを実運用できます。
• オントロジーでエンドユーザーから収集した業務データをほかのシステムに書き込み、Foundry の外部で分析できます。

エクスポートすると、データへのアクセス、保持、共有が可能な場所が増えます。エクスポートを設定する前に、宛先と受信者が承認された目的に適していることを確認し、エクスポートするフィールドを最小限に抑え、適切なアクセス制限を適用してください。目的の限定と責任ある共有に関するガイダンスについては、プライバシーと市民的自由の実践を参照してください。

Data Connection のエクスポートは、まだすべてのソースの種類でサポートされているわけではありません。利用可能かどうかを確認するには、ソースの種類の概要ドキュメントに記載されている個々のソースページを参照してください。エクスポート機能が利用可能な場合、各ソースページには、ファイルエクスポート、ストリーミングエクスポート、テーブルエクスポート、レガシーエクスポートタスクのいずれかのサポートが記載されています。たとえば、Amazon S3 ソースの種類はファイルエクスポートをサポートしていますが、BigQuery ソースにはエクスポートタスクが必要です。

ファイル、ストリーミング、テーブルのエクスポート機能をまだサポートしていないソースの一部では、レガシーエクスポートタスクの設定がサポートされている場合があります。更新されたエクスポート機能を実装しているソースの種類では、エクスポートタスクは推奨されなくなりました。

エクスポートの種類

エクスポートの動作は、宛先システムが受け入れられるデータの種類によって異なります。

ファイルエクスポート

ファイルエクスポートをサポートするシステムの例として、S3 があります。

ファイルエクスポートでは、選択した Foundry データセットのファイルを、設定した宛先に書き込みます。デフォルトでは、上流データセットで最後に正常にエクスポートされたトランザクション以降に変更されたファイルのみが書き込まれます。つまり、あるトランザクションでファイルが更新されなかった場合、下流のどのエクスポートでも、次回のスケジュール実行または手動実行でそのファイルが再エクスポートされることはありません。

データセット全体を再エクスポートする必要がある場合は、同じソースに新しいエクスポートを設定するか、エクスポートが必要なすべてのファイルが上流の変換処理で上書きされるようにしてください。

デフォルトでは、宛先にファイルがすでに存在する場合、エクスポートジョブはそのファイルをエクスポートしたデータで上書きします。この動作は、ソースの種類によって異なる場合があります。宛先システムに保存されているデータを誤って上書きしないように、Foundry からエクスポートしたデータを格納する専用のサブフォルダーを作成することを推奨します。

ストリーミングエクスポート

ストリーミングの宛先へのエクスポートでは、Foundry でエクスポートジョブが実行されている間、レコードが宛先にストリーミングされます。ジョブを停止して再起動すると、停止した位置からレコードのストリーミングが再開されます。

次のソースの種類がストリーミングエクスポートをサポートしています。

• Amazon Kinesis
• Amazon SQS
• Google Pub/Sub
• Kafka
• PostgreSQL
• Solace

Kafka にエクスポートする場合、ストリームレコードの Base64 デコードを有効にできます。Kafka エクスポートの詳細については、Kafka コネクターのドキュメントを参照してください。

ストリーミングエクスポートのリプレイ時の動作

ストリーミングの宛先へのエクスポートを設定する際は、Foundry でストリームがリプレイされたときの動作を指定する必要があります。通常、処理ロジックに破壊的変更を加えた後に、ストリームがリプレイされます。この場合、以前に処理されたレコードを、次のいずれかの方法で新しいロジックを使用して再処理する必要があります。

• リプレイされたレコードをエクスポート： ストリームがリプレイされたときに、すべてのレコードを再エクスポートします。以前にエクスポートされたレコードはすべて再びエクスポートされるため、外部システムは重複を処理できるように設定する必要があります。

• リプレイされたレコードをエクスポートしない： リプレイされたオフセットがエクスポートジョブの最新のオフセットと一致するまで、エクスポートを一時停止します。リプレイされたストリーム間でオフセットが一致する保証はないため、このオプションを選択すると、一部のレコードが欠落し、そのままエクスポートされなくなることがよくあります。

テーブルエクスポート

テーブルエクスポートは2024年6月より一般提供されていますが、一部のソースの種類では利用できない場合があります。利用可能な場合は、エクスポートタスクよりもテーブルエクスポートを推奨します。ソースの種類がテーブルエクスポートに対応しているかどうかを確認するには、Data Connection のソースページで「テーブルエクスポート」タグを探します。該当するソースの種類のドキュメントページでも確認できます。

テーブルエクスポートを使用すると、既知のスキーマを持つテーブル形式のデータを含むシステムにエクスポートできます。この場合、外部の対象スキーマは、エクスポートする Foundry データセットのスキーマと一致している必要があります。テーブルエクスポートでは、以下で説明する選択済みのテーブルエクスポートモードに従って、システム固有の INSERT 文または同等の API 呼び出しを実行し、対象テーブルに行を書き込みます。

テーブルエクスポートモード

選択したエクスポートモードは、テーブルエクスポート時にデータをどのようにエクスポートするかを指定します。以下のエクスポートモードはすべてのテーブルエクスポートで利用可能で、Data Connection のエクスポートのセットアップページでも説明されています。

レガシーエクスポートタスクでは、エクスポートモードは利用できません。

テーブルエクスポートに関する特別な考慮事項

• Foundry からエクスポートするデータセットは、列名の完全一致（大文字と小文字を区別）とデータ型を含め、外部ソースと1対1で一致する必要があります。たとえば、LONG 型の COLUMN_ABC を、COLUMN_ABC が STRING 型である宛先にエクスポートすると、実行時に失敗します。
• テーブルエクスポートをセットアップする際は、ソースで必須とされている場合、DATABASE、SCHEMA、TABLE の各項目が入力されていることを確認してください。これらの項目のいずれかが未入力の場合、エクスポートは実行時に失敗します。同様に、不要な項目を含めることでも失敗する場合があります。信頼性を最大限に高めるには、ソース探索と自動入力を使用して、ソース内の宛先を選択します。
• テーブルエクスポートで Agent ワーカーを使用する場合、Agent は Linux ホスト上で稼働している必要があります。Windows Agent ではテーブルエクスポートを実行できません。
• テーブル形式の宛先にエクスポートするデータセットの基となるファイルは、Parquet 形式である必要があります。CSV や、スキーマに対応するその他のファイル形式では、エクスポートに失敗します。
• 宛先テーブルは、ソースシステムにすでに存在している必要があります。Foundry によって自動的に作成されることはありません。
• 切り詰めを行うエクスポートモードを使用する場合、Foundry のソース設定に入力した資格情報には、外部システムのテーブルを切り詰める権限が必要です。
• Array、Map、Struct 型のエクスポートには対応していません。エクスポートするデータセットに Array、Map、または Struct 型の列が含まれている場合、エクスポートは失敗します。

テーブルエクスポートの SQL 方言

現在、テーブルエクスポートはカスタム JDBC ソースタイプに対応しています。Foundry は Oracle などの特定のシステム向けに専用のコネクターを提供していますが、カスタム JDBC 接続を介して接続を確立しない限り、テーブルエクスポートのソース機能は選択肢として表示されません。

独自のドライバー JAR ファイルを指定したカスタム JDBC ソースに対してテーブルエクスポートを実行する場合は、データのエクスポートに使用される Palantir の INSERT 文の構文が、システムで受け付けられる SQL 方言に対応していることを確認してください。

以下に、カスタム JDBC ソースへのエクスポートに Palantir が使用する INSERT 文の例を示します。現在、これらの文はカスタマイズできません。エクスポートの実行時に使用される文をシステムが受け付けない場合は、代わりにコードベースの接続オプションを使用してシステムに書き込むことができます。

また、データベースで使用されているデータ型が、以下に示す対応データ型と互換性があることを確認してください。

対応する JDBC データ型

対応する JDBC ソースのデータ型は次のとおりです。

• DECIMAL
• BIGINT
• BINARY
• BOOLEAN
• SMALLINT
• DATE
• DOUBLE
• FLOAT
• INTEGER
• VARCHAR
• TIMESTAMP

未対応の JDBC データ型

未対応の JDBC ソースのデータ型は次のとおりです。

• ARRAY
• MAP
• STRUCT
• 上記の対応データ型として記載されていないその他のデータ型（GEOGRAPHY など、あまり一般的でないデータ型）

データセットに未対応のデータ型（ARRAY、MAP、STRUCT など）の列が含まれている場合、エクスポートに失敗します。

INSERT 文の例

以下の例では、テーブルエクスポートで使用される構文について詳しく説明します。ソースシステムがこの形式の INSERT 文に対応していない場合、INSERT 文の実行時に失敗します。

ソースシステム内に target_table が存在し、Foundry 内の対応するデータセットと同じスキーマを持つと仮定します。列の名前とデータ型は、外部ソースと Foundry の間でどちらも一致している必要があります。以下は、Foundry 内のデータセットの仮の例です。

1行に対する INSERT 文は、以下の一般的な形式に従います。

INSERT INTO target_table (column1, column2, column3) VALUES ('value1', 'value2', 'value3');

ソースタイプに応じて、INSERT 文は1行または複数行のデータに適用されます。たとえば、Snowflake はバッチ挿入に対応しているため、複数行を一度に書き込むことができます。

INSERT INTO target_table (column1, column2, column3) VALUES ('value1', 'value2', 'value3'), ('value4', 'value5', 'value6'), ('value7', 'value8', 'value9');

一方、Apache Hive のようなソースではバッチ挿入への対応が限定的であり、3つの INSERT 文を順に実行してエクスポートします。

INSERT INTO target_table (column1, column2, column3) VALUES ('value1', 'value2', 'value3');
INSERT INTO target_table (column1, column2, column3) VALUES ('value4', 'value5', 'value6');
INSERT INTO target_table (column1, column2, column3) VALUES ('value7', 'value8', 'value9');

TRUNCATE TABLE 文の例

切り詰めを伴うデータセット全体などのエクスポートオプションでは、まず外部ソースの対象テーブルを切り詰めてから、INSERT 文を実行します。これを実行する実際のコードは、以下の例のようになります。

TRUNCATE TABLE table_name;
INSERT INTO target_table (column1, column2, column3) VALUES ('value1', 'value2', 'value3');

より特定の方言が必要な場合は、外部トランスフォームなど、Foundry のコードベースの接続オプションのいずれかを推奨します。

マルチスレッドのテーブルエクスポート

一部のソースタイプはマルチスレッドのテーブルエクスポートに対応しており、エクスポートのパフォーマンスを向上させることができます。マルチスレッドのエクスポートを使用する場合、Foundry は原子性を維持するために、宛先システムに一時的なステージングテーブルを作成します。データはこのステージングテーブルに並列で書き込まれ、エクスポートが完了すると最終的な宛先テーブルにコピーされます。ステージングテーブルは、エクスポートの終了後に自動的に削除されます。

マルチスレッドのエクスポートを使用する場合、資格情報には、宛先システム内のテーブルに対する CREATE、READ、UPDATE、DELETE 権限が必要です。

エクスポート用にデータを準備する

原則として、エクスポートジョブの一環としてデータ変換を実行することには対応していません。つまり、エクスポート対象として選択するデータセットまたはストリームは、フィルターの適用、ファイル名の変更やパーティションの再分割、その他のデータ変換を含め、すでに目的の形式になっている必要があります。

Pipeline Builder と Code Repositories は、データ変換パイプラインを構築するための Foundry のツールです。どちらのアプリケーションも、ジョブのスケーラビリティ、監視、バージョン管理、必要に応じて任意のロジックを記述できる柔軟性など、エクスポート用のデータを準備するために必要なツールを一通り備えています。

出力ファイルの構造を制御する

構造化データを含む Foundry データセットでは、デフォルトで元のファイルが spark/ ディレクトリに保存されます。エクスポート時にファイルを特定のディレクトリ構造（たとえば、日付パーティションごとに整理した構造）に配置する必要がある場合は、Python トランスフォームを使用して、エクスポート前に目的のレイアウトでデータセットを書き込むことができます。

以下の Python トランスフォームは、ファイルシステム API を使用して、日付に基づくパスにファイルを直接コピーします。

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

@transform(
    output_dataset=Output("/path/to/output/dataset"),
    input_dataset=Input("/path/to/input/dataset"),
)
def compute(input_dataset, output_dataset):
    dt = datetime.date.today().strftime("%Y%m%d")
    for file_status in input_dataset.filesystem().ls():
        relative_path = file_status.path[len("spark/"):] if file_status.path.startswith("spark/") else file_status.path
        with input_dataset.filesystem().open(file_status.path, "rb") as in_f:
            with output_dataset.filesystem().open(f"data_dt={dt}/{relative_path}", "wb") as out_f:
                shutil.copyfileobj(in_f, out_f)

このトランスフォームを使用すると、エクスポートされたデータセットには、spark/part-00000-xxx.parquet ではなく、data_dt=20240101/part-00000-xxx.parquet などのパスにファイルが配置されます。この変換済みデータセットをエクスポート対象のデータセットとして使用するように、ファイルエクスポートを設定します。初回のビルド後、データセットでスキーマを適用を選択すると、Foundry でテーブル形式のデータとして表示できるようになります。

エクスポートをセットアップする

ソースのエクスポートを有効にする

データをエクスポートするには、エクスポート先のソースの接続設定セクションでエクスポートを有効にする必要があります。エクスポートをサポートするソースの種類では、エクスポート設定タブが表示されます。デフォルトロールでは、次の権限が適用されます。

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

エクスポートを有効にした後、このソースへのエクスポートを許可するマーキングを指定します。エクスポート設定に含まれていないマーキングまたは組織が付いたデータは、このソースにエクスポートできません。エクスポート可能なマーキングや組織を追加するために、情報セキュリティ責任者ロールは必要ありません。詳細については、マーキング権限、組織権限、およびソース権限を確認してください。

たとえば、Palantir 組織内に機密マーキングが付いたデータセットがあるとします。このデータセットをエクスポートするには、このソースのエクスポート可能なマーキングに、機密マーキングと Palantir 組織の両方を追加する必要があります。

エクスポート可能なマーキングは、監視ビューからの Slack 通知に表示できるリソースのメタデータも制御します。

新しいエクスポートを作成する

エクスポートを作成するには、まずエクスポート先のソースの概要ページに移動します。

そのソースで初めてエクスポートをセットアップする場合は、空のテーブルとエクスポートを作成するボタンが表示されます。

エクスポートを作成を選択し、エクスポートするデータセットまたはストリームと、ソース固有のエクスポート設定オプションを選択します。これらのオプションはソースコネクターによって異なり、ドキュメント内の対応するソースの種類のページで説明されています。エクスポートするデータセットに複数のブランチが存在する場合、master ブランチのデータのみがエクスポートされます。

以下の例は、S3 コネクターのエクスポート設定インターフェースを示しています。

エクスポートを保存すると、エクスポート管理ページが表示され、次の操作ができます。

• エクスポートを手動で実行する。
• エクスポートのスケジュールを設定する。
• エクスポート履歴を表示する。
• 設定オプションを変更する。

ストリーミングエクスポートでは、実行ボタンの代わりに開始/停止ボタンを使用します。ストリーミングエクスポートにスケジュールが設定されている場合、ほかのストリームのスケジュールと同様に動作します。スケジュールがトリガーされたときにストリームが停止している場合は、自動的に再開されます。スケジュールがトリガーされたときにストリームが停止していない場合は、そのまま実行が続きます。

エクスポートをスケジュールする

エクスポートは、最新のデータを外部の宛先にエクスポートするために、定期的に実行するようスケジュールする必要があります。ストリーミングエクスポートは、開始または停止するだけでよいため、スケジュールする必要はありません。

エクスポートをスケジュールするには、そのエクスポートの概要ページに移動します。次に、スケジュールを追加を選択して、Data Lineage でエクスポートを開きます。そこで、画面の右側にある新しいスケジュールを作成するを選択し、ほかのジョブと同様に設定します。利用可能なスケジュールオプションの詳しくはこちらをご覧ください。

特定のエクスポートをトリガーするスケジュールは、以下のように、そのエクスポートの概要ページで確認できます。

エクスポート履歴

同期と同様に、エクスポートは Foundry のビルドシステムを使用してジョブとして実行されます。エクスポートの履歴表示には、関連するジョブの履歴が表示されます。ジョブ詳細セクションの右上にあるビルドレポートを表示を選択すると、各ジョブをビルドアプリケーションで開いて表示できます。

ストリーミングエクスポートの場合、エクスポート履歴には、ストリーミングエクスポートジョブが現在実行中か停止中かも表示されます。

エクスポートとエクスポートタスクの比較

テーブルエクスポートは、現在サンセット段階にあるエクスポートタスクを完全に置き換えることを目的としています。エクスポートタスクは、公開済みの製品ライフサイクルに従って、最終的に廃止されます。エクスポートタスクで提供されている機能の多くは、新しいエクスポートオプションでも利用できますが、機能や設定にはいくつかの違いがあります。

エクスポートタスクからテーブルエクスポートに移行する

エクスポートタスクから新しいエクスポートオプションへの移行は、手動で行う必要があります。以下に記載されているオプションを使用してエクスポートを設定し、新しいエクスポート処理が期待されるとおりに動作するようになったら、以前使用していたエクスポートタスクを手動で削除します。

エクスポートタスクで利用できる機能に相当するオプションが新しいエクスポートにない場合は、外部トランスフォームの使用に切り替えることを推奨します。外部トランスフォームは、外部システムとやり取りするカスタムロジックを実行するための柔軟な代替手段です。ソースの機能に関する UI の設定オプションで必要な機能をカバーできない場合に使用することを目的としています。

エクスポートとエクスポートタスクの機能比較