Amazon S3

Foundry を AWS S3 に接続して、S3 と Foundry 間でデータの読み取りと同期を行います。

サポートされる機能

セットアップ

1. Data Connection アプリケーションを開き、画面右上の + 新規ソースを選択します。
2. 利用可能なコネクターの種類から S3 を選択します。
3. 以下のセクションの情報を使用し、表示される追加設定のプロンプトに従って、コネクターのセットアップを続けます。

Foundry でのコネクターのセットアップについて詳しくはこちら。

接続情報

S3 に必要な読み取りと同期の権限

S3 バケットを対話的に探索するには、次の AWS 権限が必要です。

Copied!
1
2
3
4
5
{
    "Action": ["s3:ListBucket"],
    "Resource": ["arn:aws:s3:::path/to/bucket"],
    "Effect": "Allow",
}

S3 からのバッチ同期、仮想テーブル、メディア同期には、次の AWS 権限が必要です。

Copied!
1
2
3
4
5
{
    "Action": ["s3:GetObject"],
    "Resource": ["arn:aws:s3:::path/to/bucket/*"],
    "Effect": "Allow",
}

S3 でのバケットポリシーの設定方法の詳細については、Amazon S3 のポリシーと権限に関する AWS の公式ドキュメント ↗ を参照してください。

S3 プロキシ設定（Agent ベースの接続）

Data Connection の Agent を使用して S3 に接続する場合、次の2つの方法でプロキシ設定を定義できます。

• ソース設定：以下のテーブルに示すとおり、ソース設定で各プロキシ設定を直接定義します。
• Agent のシステムプロパティ：フォールバックとして、Agent のシステムプロパティでプロキシ設定を行えます。そのためには、Agent の詳細設定に適切な JVM 引数（たとえば -Dhttps.proxyHost=example.proxy.com）を含めます。

STS ロール設定

STS ロール設定により、S3 から読み取る際に AWS Security Token Service ↗ を使用してロールを引き受けることができます。

クラウド ID 設定

クラウド ID 認証により、Foundry は AWS インスタンス内のリソースにアクセスできます。クラウド ID は、Control Panel でエンロールメントレベルで設定および管理されます。クラウド ID を設定する方法を参照してください。

クラウド ID 認証を使用する場合、資格情報セクションにロール ARN が表示されます。資格情報の選択肢としてクラウド ID を選択した後、次の設定も行う必要があります。

1. 対象の Amazon AWS アカウントで Identity and Access Management (IAM) ロールを設定します。
2. 接続する S3 バケットへのアクセス権を IAM ロールに付与します。通常はバケットポリシー ↗を使用して行えます。
3. S3 ソースの設定詳細で、Security Token Service (STS) ロール ↗設定に IAM ロールを追加します。Foundry のクラウド ID の IAM ロールは、S3 にアクセスする際に AWS アカウントの IAM ロール ↗を引き受けようとします。
4. クラウド ID の IAM ロールが対象の AWS アカウントの IAM ロールを引き受けられるように、対応する信頼ポリシーを設定します ↗。

仮想テーブル

このセクションでは、S3 ソースの仮想テーブルの使用に関する補足情報を説明します。Foundry データセットへの同期には、このセクションは適用されません。

次のテーブルに、S3 でサポートされている仮想テーブルの機能を示します。

S3 に保存されたテーブルを入力または出力として使用できる、サポート対象の Foundry ワークフローについては、仮想テーブルのドキュメントを参照してください。

ソース設定の要件

仮想テーブルを登録する際は、次のソース設定の要件に注意してください。

• Foundry ワーカーのソースを使用する必要があります。仮想テーブルでは、Agent ワーカー接続の使用はサポートされていません。
• 接続情報のネットワーク接続セクションの説明に従って、双方向接続と許可リスト登録が確立されていることを確認してください。
• Code Repositories で仮想テーブルを使用する場合、追加で必要なソース設定の詳細については、仮想テーブルのドキュメントを参照してください。
• バケットに . が含まれる場合は、パス形式アクセスを有効にして、適切なエグレスポリシーを設定する必要があります。

詳細については、上記の接続情報セクションを参照してください。

Delta

仮想テーブルを元にするパイプラインでインクリメンタル処理のサポートを有効にするには、ソースの Delta テーブルで Change Data Feed ↗ が有効になっていることを確認してください。Python トランスフォームの current および added 読み取りモードがサポートされています。Python トランスフォームでは、_change_type、_commit_version、_commit_timestamp 列が利用可能になります。

Iceberg

Apache Iceberg テーブルを元にする仮想テーブルを読み込むには、Iceberg カタログが必要です。Iceberg カタログの詳細については、Apache Iceberg のドキュメント ↗を参照してください。ソースに登録されるすべての Iceberg テーブルは、同じ Iceberg カタログを使用する必要があります。

デフォルトでは、S3 内の Iceberg メタデータファイルを使用してテーブルが作成されます。テーブルを登録する際は、これらのメタデータファイルの場所を示す warehousePath を指定する必要があります。

テーブルが S3 に保存されている場合、AWS Glue ↗ を Iceberg カタログとして使用できます。この統合の詳細については、AWS Glue のドキュメント ↗を参照してください。ソースに設定された資格情報には、AWS Glue データカタログへのアクセス権が必要です。AWS Glue は、ソースの接続情報タブで設定できます。このソースに登録されたすべての Iceberg テーブルは、自動的に AWS Glue をカタログとして使用します。テーブルは database_name.table_name という命名パターンで登録してください。

Databricks で Delta Universal Format (UniForm) を使用する場合、Unity Catalog ↗ を Iceberg カタログとして使用できます。この統合の詳細については、Databricks のドキュメント ↗を参照してください。AWS Glue と同様に、カタログはソースの接続情報タブで設定できます。Unity Catalog に接続するには、エンドポイントと個人用アクセストークンを指定する必要があります。テーブルは catalog_name.schema_name.table_name という命名パターンで登録してください。

Polaris ↗ は、S3 に保存されたテーブルの Iceberg カタログとして使用できます。ほかのカタログと同様に、Polaris はソースの接続情報タブで設定できます。

インクリメンタル処理のサポートは、Iceberg のインクリメンタル読み取り ↗に依存しており、現在は APPEND のみです。Python トランスフォームの current および added 読み取りモードがサポートされています。

Parquet

Parquet を使用する仮想テーブルは、スキーマ推論に依存します。スキーマの特定には、最大100個のファイルが使用されます。

データを S3 にエクスポートする

S3 にエクスポートするには、まず S3 コネクターのエクスポートを有効にします。次に、新しいエクスポートを作成します。

S3 へのエクスポートに必要な権限

データを S3 にエクスポートするには、次の AWS 権限が必要です。

Copied!
1
2
3
4
5
{
    "Action": ["s3:PutObject"],
    "Resource": ["arn:aws:s3:::path/to/bucket/*"],
    "Effect": "Allow",
}

S3 のバケットポリシーを設定する方法の詳細については、Amazon S3 のポリシーとアクセス許可に関する AWS 公式ドキュメント ↗を参照してください。

Iceberg BYOB ストレージに必要な権限

S3 バケットを Foundry Iceberg テーブルの bring-your-own-bucket (BYOB) ストレージの場所として使用する場合、S3 バケットと、その暗号化に使用する KMS キーに対して、次の AWS 権限が必要です。

Copied!
1
2
3
4
5
6
7
8
9
10
11
12
13
{
    "Action": [
        "s3:DeleteObject",
        "s3:GetObject",
        "s3:ListBucket",
        "s3:PutObject"
    ],
    "Resource": [
        "arn:aws:s3:::bucket-name",
        "arn:aws:s3:::bucket-name/*"
    ],
    "Effect": "Allow"
}

さらに、次の KMS 権限が必要です。

Copied!
1
2
3
4
5
6
7
8
9
{
    "Action": [
        "kms:Decrypt",
        "kms:Encrypt",
        "kms:GenerateDataKey"
    ],
    "Resource": ["arn:aws:kms:region:account-id:key/key-id"],
    "Effect": "Allow"
}

資格情報の払い出しには、sts:GetFederationToken 権限も必要です。セットアップ手順の全容については、BYOB ストレージのドキュメントを参照してください。

エクスポート設定のオプション

エクスポートタスク（レガシー）

S3 エクスポートタスクの設定

エクスポートタスクを使用して Foundry データセットから S3 バケットにファイルをエクスポートするには、ソースの種類が s3-direct である必要があります。タスクの種類は export-s3-task です。

基本設定

Copied!
1
2
type: export-s3-task
directoryPath: /some/directory

詳細設定オプション

設定例

再試行を伴うインクリメンタルエクスポート：

Copied!
1
2
3
4
type: export-s3-task
directoryPath: /exports/data
incrementalType: incremental
retriesPerFile: 2

カスタムバケットポリシーを使用した並列エクスポート：

Copied!
1
2
3
4
5
6
type: export-s3-task
directoryPath: /exports/data
useMultipart: false
threads: 4
bucketPolicy: BucketOwnerFullControl
setBucketPolicy: true

Spark の接頭辞を削除するためのパスの書き換え：

Copied!
1
2
3
4
type: export-s3-task
directoryPath: /exports/data
rewritePaths:
  "^spark/(.*)": "$1"

コードで S3 ソースを使用する

Python トランスフォームのコードリポジトリから S3 バケットに接続して、データセットをエクスポートしたり、外部トランスフォームを使用してファイルを処理したりできます。

S3 バケットにファイルを書き込む

以下の例では、トランスフォームジェネレーターを使用して、データセット RID の集合を S3 バケットにエクスポートする方法を示します。

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
# この例では、AWS セッション資格情報を使用して、データセットのリストを S3 バケットにエクスポートする方法を示します。
# エクスポートしたデータセットごとに、データセットの RID、エクスポートのタイムスタンプ、エクスポートした
# ファイル数、および S3 の宛先パスを含む出力ログが生成されます。

from pyspark.sql import functions as F, types as T
from transforms.api import Input, Output, transform, Transform
from transforms.external.systems import external_systems, Source
from typing import List
from datetime import datetime
import boto3
import logging

log = logging.getLogger(__name__)

S3_BUCKET_REGION = "<region>"
S3_BUCKET_NAME = "<name>"
S3_ENDPOINT_URL = "<endpoint_url>"
S3_DEST_PATH = "<s3_dest_path>"

S3_SOURCE_RID = "<s3_source_rid>"

LOG_OUTPUT_PATH = "<log_output_path>"

DATASET_RID = "<dataset_rid>"

@external_systems(
    s3_source=Source(S3_SOURCE_RID)
)
@transform(
    export_log=Output(f"{LOG_OUTPUT_PATH}/export_{DATASET_RID}_log"),
    input_dataset=Input(f"{DATASET_RID}")
)
def compute_function(ctx, input_dataset, s3_source, export_log):
    # 1. S3 バケットの資格情報を取得する
    refreshable_creds = s3_source.get_session_credentials()
    creds = refreshable_creds.get()

    # 2. 現在のデータセットの S3 キーを構築する
    s3_dataset_key = f"{S3_DEST_PATH}/{DATASET_RID}/"

    # 3. boto3 クライアントをセットアップする
    # 注：資格情報の TTL（約1時間）を超える可能性がある大規模なエクスポートでは、
    # 資格情報を定期的に更新するか、
    # boto3 の RefreshableCredentials パターンを実装することを検討してください。
    s3_client = boto3.client(
        "s3",
        aws_access_key_id=creds.access_key_id,
        aws_secret_access_key=creds.secret_access_key,
        region_name=S3_BUCKET_REGION,
        endpoint_url=S3_ENDPOINT_URL,
    )

    # データセットのファイルを S3 に書き込むヘルパー関数
    def write_file(row):
        try:
            with input_dataset.filesystem().open(row.path, "rb") as in_f:
                s3_key = s3_dataset_key + row.path
                s3_client.upload_fileobj(in_f, S3_BUCKET_NAME, s3_key)

        except Exception as e:
            log.error(f"Exception when uploading {DATASET_RID} to S3: {e}")
            raise

    # 4. データセットのファイルを S3 に書き込む
    result = input_dataset.filesystem().files()
    result.foreach(write_file)

    # 5. 出力するエクスポートログのスキーマを定義する
    schema = T.StructType([
        T.StructField("dataset_rid", T.StringType(), False),
        T.StructField("export_timestamp", T.TimestampType(), False),
        T.StructField("files_uploaded", T.IntegerType(), False),
        T.StructField("s3_dest_path", T.StringType(), False)
    ])
    file_count = result.count()
    output_df = ctx.spark_session.createDataFrame([(DATASET_RID, datetime.now(), file_count, s3_dataset_key)], schema)
    export_log.write_dataframe(output_df)