データパイプライン宛先としてS3 Icebergを設定する
データパイプラインの宛先としてS3 Icebergをセットアップします。このコネクションにより、WorkatoはソースアプリケーションのデータをAmazon S3内のIcebergテーブルにレプリケートできます。これらのテーブルはAWS Glue Data Catalog、またはSnowflake Horizon Catalog、Apache Polaris、Databricks Unity CatalogなどのIceberg RESTカタログを基盤とします。
サポートされている機能
S3 Icebergをパイプライン宛先として使用する場合、次の機能がサポートされます:
- ソーススキーマに基づく宛先テーブルの自動作成。
- 明示的なフィールドマッピングなしのフィールドレベルのデータレプリケーション。
- AWS Glue Data CatalogまたはIceberg RESTカタログ（Snowflake Horizon Catalog、Apache Polaris、Databricks Unity Catalog、またはセルフホスト型iceberg-restサーバー）の選択。
- RESTカタログ向けの静的キーまたはベンドされた認証情報によるAWS認証、およびBearer tokenまたはOAuth 2.0によるカタログ認証。
前提条件
次の設定とアクセス権が必要です:
- Icebergデータファイルを保存するAmazon S3バケットと、そのバケットが配置されているAWSリージョン
- Icebergカタログ。 AWS Glue Data Catalog、またはSnowflake Horizon Catalog、Apache Polaris、Databricks Unity Catalog、またはセルフホスト型iceberg-restサーバーなどのRESTカタログ
- 静的アクセスキーIDとシークレットとして、またはRESTカタログによって発行されるベンドされた認証情報を通じて提供されるAWS認証情報。必要な権限については、S3 Iceberg用のAWS認証情報を作成するを参照してください。
- RESTカタログを使用する場合は、サポートされているカタログ認証方式
S3 Iceberg用のAWS認証情報を作成する
カタログタイプとしてAWS Glueを使用する場合、またはRESTカタログでAWS認証方式としてStatic keysを使用する場合は、専用のIAMユーザーを作成します。ベンドされた認証情報を使用する場合は、このセクションをスキップしてください。代わりにS3 IcebergはRESTカタログから一時的なテーブル単位の認証情報をリクエストします。
次の権限を持つIAMポリシーをAWSで作成します:
{
"Version": "2012-10-17",
"Statement": [
{
"Effect": "Allow",
"Action": [
"s3:ListBucket*",
"s3:GetBucketLocation",
"s3:GetObject*",
"s3:PutObject",
"s3:PutObjectAcl",
"s3:DeleteObject"
],
"Resource": [
"arn:aws:s3:::<your-bucket>/*",
"arn:aws:s3:::<your-bucket>"
]
},
{
"Effect": "Allow",
"Action": [
"glue:TagResource",
"glue:UnTagResource",
"glue:BatchCreatePartition",
"glue:BatchDeletePartition",
"glue:BatchDeleteTable",
"glue:BatchGetPartition",
"glue:CreateDatabase",
"glue:CreateTable",
"glue:CreatePartition",
"glue:DeletePartition",
"glue:DeleteTable",
"glue:GetDatabase",
"glue:GetPartition",
"glue:GetPartitions",
"glue:GetTable",
"glue:GetTables",
"glue:UpdateDatabase",
"glue:UpdatePartition",
"glue:UpdateTable"
],
"Resource": ["*"]
},
{
"Effect": "Allow",
"Action": [
"athena:StartQueryExecution",
"athena:GetQueryExecution",
"athena:GetQueryResults",
"athena:StopQueryExecution",
"athena:GetWorkGroup",
"athena:ListWorkGroups"
],
"Resource": ["*"]
}
]
}S3ステートメントのResource値の範囲をウェアハウスバケットに限定します。 s3アクションはすべてのカタログタイプに適用されます。 glueおよびathenaアクションは、AWS Glueを使用する場合にのみ適用されます。
IAMユーザーを作成し、前のステップのポリシーをアタッチします。
ユーザーのアクセスキーを生成します。コネクションを設定する際は、生成されたaccess key IDとsecret access keyを使用します。
RESTカタログの場合は、IAMポリシーではなくカタログ自体を通じてアクセスを付与します。たとえば、Snowflake Horizon Catalogでは、CREATE ICEBERG TABLE権限と、S3バケット用に設定された外部ボリュームの使用権限を持つロールに加えて、Bearer tokenまたはOAuth 2.0 credential値として使用するプログラムによるアクセストークンが必要です。
S3 Icebergに接続する
データパイプライン宛先としてS3 Icebergに接続するには、次の手順を完了します:
S3 Icebergに接続する
作成 > コネクションを選択するか、Cを2回押します。
S3 Icebergを検索し、New connectionページで選択します。
コネクション名フィールドに名前を入力します。
ロケーションドロップダウンメニューを使用して、コネクションを保存するプロジェクトを選択します。
該当する場合は、S3バケットとGlueカタログのAWS regionを入力します。 Workatoでは、ベンドされた認証情報を使用する場合でもこの値が必要です。
Warehouseフィールドに値を入力します:
- AWS Glue: Icebergデータファイルが保存されるS3 URI（
s3://my-bucket/warehouseなど）。 - REST catalogs: カタログ固有の値。 Snowflakeの場合は、
MY_DATABASEなどのデータベース名を入力します。 Databricks Unity Catalogまたはセルフホスト型iceberg-restカタログの場合は、S3 URIを入力します。
Catalog typeドロップダウンメニューを使用して、宛先の基盤となるIcebergカタログを選択し、選択内容に応じて該当する手順を完了します:
接続をクリックして、コネクションを検証および確立します。
宛先アクションの設定
パイプラインを開始する前に、宛先ウェアハウスとカタログが新しく作成され、空であることを確認します。これにより、初回同期中のエラーを防ぎ、パイプラインが競合なしで送信先テーブルを作成できるようになります。
宛先アプリのターゲットテーブルにデータをロードアクションをクリックします。このアクションでは、パイプラインが宛先でデータをレプリケートする方法を定義します。
利用可能な宛先アプリのリストからS3 Icebergを選択します。
このパイプラインで使用するS3 Icebergコネクションを選択します。または、+ 新規コネクションをクリックして新しいコネクションを作成します。
Load data to target table in destination appアクションは、ソースからS3 Icebergへオブジェクトスキーマを自動的にレプリケートします。明示的なフィールドマッピングは不要です。
Namespaceフィールドに値を入力します。これにより、ターゲットテーブルが作成されるIceberg名前空間が設定されます:
- AWS Glue:
sales_pipelineなどのGlueデータベース名。 Glueデータベース名は1～255文字である必要があります。完全な命名パターンについては、AWS Glue Table APIリファレンスを参照してください。 - REST catalogs: カタログ固有の値。 Snowflakeの場合は、ウェアハウスとして指定されたデータベース内のスキーマ名（
SALES_PIPELINE_SCHEMAなど）を入力します。命名制限はカタログによって異なるため、カタログベンダーの識別子ルールを参照してください。
S3 Iceberg宛先の設定
このフィールドを空白のままにすると、Workatoはworkato名前空間の下にテーブルを作成します。
保存を選択してパイプラインを保存します。
最終更新日: