# Amazon Data Firehose を使用したテーブルへのデータのストリーミング
Amazon Data Firehose は、Amazon S3、Amazon Redshift、Amazon OpenSearch Service、Splunk、Apache Iceberg テーブル、カスタム HTTP エンドポイント、またはサポートされているサードパーティーのサービスプロバイダーが所有する HTTP エンドポイントなどのターゲットにリアルタイムの[ストリーミングデータ](https://aws.amazon.com//streaming-data/)を配信するためのフルマネージドサービスです。Amazon Data Firehose を使用すると、アプリケーションを記述したり、リソースを管理したりする必要はありません。Firehose にデータを送信するデータプロデューサーを作成すると、それにより、指定した送信先にデータが自動配信されます。データを配信前に変換するように、Firehose を設定することもできます。Amazon Data Firehose の詳細については、「[Amazon Data Firehose とは](https://docs.aws.amazon.com//firehose/latest/dev/what-is-this-service.html)」を参照してください。
S3 テーブルバケット内のテーブルへの Firehose ストリーミングを設定するには、次の手順に従います。
1. [テーブルバケットを AWS 分析サービスと統合します](s3-tables-integrating-aws.md)。
1. S3 テーブルにデータを配信するように Firehose を設定します。そのためには、[Firehose がテーブルにアクセスできるように AWS Identity and Access Management (IAM) サービスロールを作成します](#firehose-role-s3tables)。
1. Firehose サービスロールにテーブルまたはテーブルの名前空間に対する明示的なアクセス許可を付与します。詳細については、「[必要なアクセス許可を付与する](https://docs.aws.amazon.com/firehose/latest/dev/apache-iceberg-prereq.html#s3-tables-prerequisites)」を参照してください。
1. [テーブルにデータをルーティングする Firehose ストリームを作成します。](#firehose-stream-tables)
## Firehose が S3 テーブルを送信先として使用するロールの作成
Firehose には、AWS Glue テーブルにアクセスし、S3 テーブルにデータを書き込むための特定のアクセス許可を持つ IAM [サービスロール](https://docs.aws.amazon.com//IAM/latest/UserGuide/id_roles_create_for-service.html)が必要です。Firehose ストリームを作成する際に、この IAM ロールが必要です。
1. IAM コンソール ([https://console.aws.amazon.com/iam/](https://console.aws.amazon.com/iam/)) を開きます。
1. 左のナビゲーションペインの **[ポリシー]** を選択します。
1. **[ポリシーを作成]** を選択し、ポリシーエディタで **[JSON]** を選択します。
1. データカタログ内のすべてのデータベースとテーブルにアクセス許可を付与する次のインラインポリシーを追加します。必要に応じて、特定のテーブルとデータベースにのみ許可を付与できます。このポリシーを使用するには、`{{user input placeholders}}` をユーザー自身の情報に置き換えます。
------
#### [ JSON ]
****
```
{
"Version":"2012-10-17",
"Statement": [
{
"Sid": "S3TableAccessViaGlueFederation",
"Effect": "Allow",
"Action": [
"glue:GetTable",
"glue:GetDatabase",
"glue:UpdateTable"
],
"Resource": [
"arn:aws:glue:{{us-east-1}}:{{111122223333}}:catalog/s3tablescatalog/*",
"arn:aws:glue:{{us-east-1}}:{{111122223333}}:catalog/s3tablescatalog",
"arn:aws:glue:{{us-east-1}}:{{111122223333}}:catalog",
"arn:aws:glue:{{us-east-1}}:{{111122223333}}:database/*",
"arn:aws:glue:{{us-east-1}}:{{111122223333}}:table/*/*"
]
},
{
"Sid": "S3DeliveryErrorBucketPermission",
"Effect": "Allow",
"Action": [
"s3:AbortMultipartUpload",
"s3:GetBucketLocation",
"s3:GetObject",
"s3:ListBucket",
"s3:ListBucketMultipartUploads",
"s3:PutObject"
],
"Resource": [
"arn:aws:s3:::{{error delivery bucket}}",
"arn:aws:s3:::{{error delivery bucket}}/*"
]
},
{
"Sid": "RequiredWhenUsingKinesisDataStreamsAsSource",
"Effect": "Allow",
"Action": [
"kinesis:DescribeStream",
"kinesis:GetShardIterator",
"kinesis:GetRecords",
"kinesis:ListShards"
],
"Resource": "arn:aws:kinesis:{{us-east-1}}:{{111122223333}}:stream/{{stream-name}}"
},
{
"Sid": "RequiredWhenDoingMetadataReadsANDDataAndMetadataWriteViaLakeformation",
"Effect": "Allow",
"Action": [
"lakeformation:GetDataAccess"
],
"Resource": "*"
},
{
"Sid": "RequiredWhenUsingKMSEncryptionForS3ErrorBucketDelivery",
"Effect": "Allow",
"Action": [
"kms:Decrypt",
"kms:GenerateDataKey"
],
"Resource": [
"arn:aws:kms:{{us-east-1}}:{{111122223333}}:key/{{KMS-key-id}}"
],
"Condition": {
"StringEquals": {
"kms:ViaService": "s3.{{us-east-1}}.amazonaws.com"
},
"StringLike": {
"kms:EncryptionContext:aws:s3:arn": "arn:aws:s3:::{{error delivery bucket}}/prefix*"
}
}
},
{
"Sid": "LoggingInCloudWatch",
"Effect": "Allow",
"Action": [
"logs:PutLogEvents"
],
"Resource": [
"arn:aws:logs:{{us-east-1}}:{{111122223333}}:log-group:{{log-group-name}}:log-stream:{{log-stream-name}}"
]
},
{
"Sid": "RequiredWhenAttachingLambdaToFirehose",
"Effect": "Allow",
"Action": [
"lambda:InvokeFunction",
"lambda:GetFunctionConfiguration"
],
"Resource": [
"arn:aws:lambda:{{us-east-1}}:{{111122223333}}:function:{{function-name}}:{{function-version}}"
]
}
]
}
```
------
このポリシーには、Kinesis Data Streams へのアクセス、Lambda 関数の呼び出し、AWS KMS キーへのアクセスを許可するステートメントが含まれています。これらのリソースを使用しない場合は、それぞれのステートメントを削除できます。
エラーログ記録が有効になっている場合、Firehose はデータ配信エラーを CloudWatch ロググループとストリームにも送信します。このためには、ロググループとログストリーム名を設定する必要があります。ロググループとログストリーム名については、「[Monitor Amazon Data Firehose Using CloudWatch Logs](https://docs.aws.amazon.com//firehose/latest/dev/controlling-access.html#using-iam-iceberg)」を参照してください。
1. ポリシーを作成したら、**[信頼されたエンティティタイプ]** として **[AWS サービス]** を使用して IAM ロールを作成します。
1. **[サービスまたはユースケース]** で、**[Kinesis]** を選択します。**[ユースケース]**で、**[Kinesis Firehose]** を選択します。
1. **[次へ]** を選択して、以前に作成したポリシーを選択します。
1. ロールに名前を付けます。ロールの詳細を確認し、**[ロールの作成]** を選択します。ロールには、以下の信頼ポリシーがあります。
------
#### [ JSON ]
****
```
{
"Version":"2012-10-17",
"Statement": [
{
"Effect": "Allow",
"Action": [
"sts:AssumeRole"
],
"Principal": {
"Service": [
"firehose.amazonaws.com"
]
}
}
]
}
```
------
## S3 テーブルへの Firehose ストリームの設定
次の手順は、コンソールを使用して S3 テーブルにデータを配信するように Firehose ストリームを作成する方法を示しています。S3 テーブルへの Firehose ストリームを設定するには、次の前提条件が必要です。
**前提条件**
+ [テーブルバケットを AWS 分析サービスと統合します](s3-tables-integrating-aws.md)。
+ [名前空間を作成します。](s3-tables-namespace-create.md).
+ [テーブルを作成する](s3-tables-create.md).
+ [Firehose が S3 テーブルにアクセスするためのロール](#firehose-role-s3tables)を作成します。
+ テーブルにアクセスするために作成した Firehose サービスロールに[必要なアクセス許可を付与](https://docs.aws.amazon.com/firehose/latest/dev/apache-iceberg-prereq.html#s3-tables-prerequisites)します。
ストリームを設定するときに Firehose にルーティング情報を提供するには、データベース名とその名前空間内のテーブル名として名前空間を使用します。これらの値は、Firehose ストリーム設定の一意のキーセクションで使用して、データを 1 つのテーブルにルーティングできます。この値を使用して、JSON クエリ式を使用してテーブルにルーティングすることもできます。詳細については、「[Route incoming records to a single Iceberg table](https://docs.aws.amazon.com/firehose/latest/dev/apache-iceberg-format-input-record.html)」を参照してください。
**S3 テーブルへの Firehose ストリームを設定するには (コンソール)**
1. Firehose コンソール ([https://console.aws.amazon.com/firehose/](https://console.aws.amazon.com/firehose/)) を開きます。
1. **[Firehose ストリームを作成]** を選択します。
1. **[ソース]** で、以下のいずれかのオプションを選択します。
+ Amazon Kinesis Data Streams
+ Amazon MSK
+ ダイレクト PUT
1. **[送信先]** で、**[Apache Iceberg テーブル]** を選択します。
1. **Firehose ストリーム名**を入力します。
1. **[ソース設定]** を構成します。
1. **[送信先の設定]** では、自分のアカウントのテーブルにストリーミングする場合は **[現在のアカウント]** を選択し、別のアカウントのテーブルにストリーミングする場合は **[クロスアカウント]** を選択します。
+ **[現在のアカウント]** のテーブルには、**[カタログ]** ドロップダウンから [S3 Tables カタログ] を選択します。
+ **[クロスアカウント]** のテーブルには、別のアカウントのストリーミング先のカタログの **[カタログ ARN]** を入力します。
1. **[一意のキー設定]**、JSONQuery 式、または Lambda 関数を使用して、データベース名とテーブル名を設定します。詳細については、「*Firehose デベロッパーガイド*」の「[着信レコードを単一の Iceberg テーブルにルーティングする](https://docs.aws.amazon.com/firehose/latest/dev/apache-iceberg-format-input-record.html)」および「[着信レコードを異なる Iceberg テーブルにルーティングする](https://docs.aws.amazon.com//firehose/latest/dev/apache-iceberg-format-input-record-different.html)」を参照してください。
1. **[バックアップ設定]** で、**[S3 バックアップバケット]** を指定します。
1. **[詳細設定]** の **[既存の IAM ロール]** で、Firehose 用に作成した IAM ロールを選択します。
1. **[Firehose ストリームを作成]** を選択します。
ストリームに対して設定できるその他の設定の詳細については、「*Amazon Firehose デベロッパーガイド*」の「[Firehose ストリームを設定する](https://docs.aws.amazon.com/firehose/latest/dev/apache-iceberg-stream.html)」を参照してください。