S3 Tablesを利用するAWSアーキテクチャ構築：Data Firehose


# はじめに
本記事は、以下記事の続きです。

https://qiita.com/tamabe/private/533d1b812de4adf4ec89

今回は、Data Firehoseストリームを設定してS3テーブルにデータを配信します。
以下AWS公式記事の手順を参考にしています。

https://docs.aws.amazon.com/AmazonS3/latest/userguide/s3-tables-integrating-firehose.html


# 構成図
最終的な構成図は以下です。
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/618752/a84d06d7-df6d-bf42-4ae9-0afb83a56f73.png)


本記事では **FirehoseからS3 Tablesにデータ配信する** 部分を実装します。



# 実装手順
以下ステップで実装していきます。
ステップ3～5では、権限管理でLake Formationも出てきます。
本記事では **3. Firehoseからデータ送信** のステップを行っていきます。

1. [S3テーブルバケットを作成](https://qiita.com/tamabe/items/8c459d2434f20191ac3e)
1. [EMRからスキーマ作成](https://qiita.com/tamabe/items/533d1b812de4adf4ec89)
2. [Firehoseからデータ送信](https://qiita.com/tamabe/items/d590444037e6d29f836e)
2. [Athenaからクエリ実行](https://qiita.com/tamabe/items/1c35ced1fee7bad9dc6e)
2. [QuickSightからデータセット接続](https://qiita.com/tamabe/items/1c7e299c13e92a3a61d8)





# Firehoseからデータ送信手順
以下ステップで行います。

1. [Firehose用IAMロール ポリシー設定](#firehose用iamロール-ポリシー設定)
1. [Lake Formationからリソースリンク作成](#lake-formationからリソースリンク作成)
2. [Lake Formationから権限設定](#lake-formationから権限設定)
2. [Firehoseストリーム作成](#firehoseストリーム作成)
2. [テストデータ送信](#テストデータ送信)



# Firehose用IAMロール ポリシー設定
まず、Firehose用のカスタムポリシーを作成します。
コンソール操作：
「IAMコンソール」 → 「ポリシー」 → 「ポリシーの作成」 → 「ポリシーエディタ: JSON」

以下JSONを貼り付け → 次へ → 任意のポリシー名設定 → ポリシーの作成
※アカウントIDやS3バケット名(Firehoseストリーム作成ステップの「その他設定」で登場)は環境に合わせて修正ください。
```json:Firehose用インラインポリシー
{
    "Version": "2012-10-17",
    "Statement": [
        {
            "Sid": "S3TableAccessViaGlueFederation",
            "Effect": "Allow",
            "Action": [
                "glue:GetTable",
                "glue:GetDatabase",
                "glue:UpdateTable"
            ],
            "Resource": [
                "arn:aws:glue:ap-northeast-1:アカウントID:catalog/s3tablescatalog/*",
                "arn:aws:glue:ap-northeast-1:アカウントID:catalog/s3tablescatalog",
                "arn:aws:glue:ap-northeast-1:アカウントID:catalog",
                "arn:aws:glue:ap-northeast-1:アカウントID:database/*",
                "arn:aws:glue:ap-northeast-1:アカウントID:table/*/*"
            ]
        },
        {
            "Sid": "S3DeliveryErrorBucketPermission",
            "Effect": "Allow",
            "Action": [
                "s3:AbortMultipartUpload",
                "s3:GetBucketLocation",
                "s3:GetObject",
                "s3:ListBucket",
                "s3:ListBucketMultipartUploads",
                "s3:PutObject"
            ],
            "Resource": [
                "arn:aws:s3:::FirehoseストリームのS3バックアップバケットで指定するS3バケット名",
                "arn:aws:s3:::FirehoseストリームのS3バックアップバケットで指定するS3バケット名/*"
            ]
        },
            {
                "Sid": "RequiredWhenDoingMetadataReadsANDDataAndMetadataWriteViaLakeformation",
                "Effect": "Allow",
                "Action": [
                    "lakeformation:GetDataAccess"
                ],
                "Resource": "*"
            },
        {
            "Sid": "LoggingInCloudWatch",
            "Effect": "Allow",
            "Action": [
                "logs:PutLogEvents"
            ],
            "Resource": [
                "arn:aws:logs:ap-northeast-1:アカウントID:log-group:*"
            ]
        }
    ]
}
```

上記設定したら、以下のようなポリシーが許可されていることを確認できます。
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/618752/14709b71-5541-000a-dba0-4e74b8ae83a0.png)


次に、Firehose用のサービスロールを作成します。
コンソール操作：
「IAMコンソール」 → 「ロール」 → 「ロールを作成」

今回、「AWSのサービス」 → 「kinesis」を選択してロールを作成しました。
※以前kinesis Data Firehoseという名前だった名残？で、「kinesis」で検索します。
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/618752/4aa1d209-2da1-0e96-0f36-ae86d94de1a6.png)


「許可ポリシー」に、先ほど作成したFirehose用ポリシーを選択して「次へ」

任意のロール名を設定 → 「ロールを作成」






# Lake Formationからリソースリンク作成
Lake Formationで、S3テーブルバケット名前空間へのリソースリンクを作成します。
※作成したリソースリンクは、Firehoseストリームの宛先として指定することになります。

コンソールやAWS CLIからも作成可能です。
※リソースリンク名は、「大文字、小文字、アンダースコア (_) 」のみ使用可能[^5]のようです。
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/618752/3c5d88ec-aa65-8a5a-95eb-7d7d58398094.png)


## AWS CLIからリソースリンク作成
```cli:
aws glue create-database --region ap-northeast-1 --catalog-id "アカウントID" --database-input \
'{
  "Name": "リソースリンク名(大文字、小文字、アンダースコア (_) のみ)",
  "TargetDatabase": {
    "CatalogId": "アカウントID:s3tablescatalog/S3テーブルバケット名",
    "DatabaseName": "S3テーブルバケット名前空間名"
  },
  "CreateTableDefaultPermissions": []
}'
```


## コンソールからリソースリンク作成
コンソールから操作：
Lake Formationコンソール → Databasesタブ → Create database

以下のように設定し、作成します。
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/618752/f0018924-1def-1b67-e0ae-86d6eef5f52f.png)





## リソースリンク作成結果
作成完了したら、以下のようにDatabasesの一覧に表示されます。
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/618752/7f3229b0-1212-298a-c6ea-fabbe84ce6b6.png)





# Lake Formationから権限設定
Data Firehoseは、リソースリンク経由でデータ配信を行います。
Lake Formationにて、Data Firehose用ロールに、リソースリンク経由のデータ配信ができるように権限を付与します。
以下2つの権限を付与します[^12]
 - リソースリンクへのDiscribe権限
 - テーブルへのALL権限

## リソースリンクへのDescribe権限付与
コンソールから操作：
Lake Formationコンソール → Databasesタブ → 作成したリソースリンク選択 → Actions → Grant

以下の通り、Data Firehose用ロールに、リソースリンクへのDescribe権限を付与します。
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/618752/eb6c5cab-3f11-07a5-dcfe-fd50528bd714.png)





## テーブルへの権限付与
コンソールから操作：
Lake Formationコンソール → Databasesタブ → 作成したリソースリンク選択 → Actions → Grant on target

以下の通り、Data Firehose用ロールに、S3テーブルバケットの指定テーブルへのALL(Super)権限を付与します。
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/618752/ccd5570e-cae1-7797-8082-952d43317617.png)

これで、FirehoseストリームがS3テーブルバケットの指定テーブルにデータ配信するための権限設定が完了しました。


# Firehoseストリーム作成
ここから、Firehoseストリームを作成していきます。

コンソールから操作：
Data Firehoseコンソール → Firehoseストリームを作成

## ソース 送信先の設定
 - ソース: 「Direct PUT」を選択します。
今回は、Data Firehoseストリームで処理するデータのソースとして後述するKDGを使うため
 - 送信先: 「**Apache Icebergテーブル**」を選択します。

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/618752/879d2dba-353e-0c05-45d4-94e6c3c31176.png)



## 一意のキー設定
一意のキー設定では、以下のようにJSON定義を設定します[^13]
```json:一意のキー設定
[
  {
    "DestinationDatabaseName": "リソースリンク名",
    "DestinationTableName": "S3テーブルバケットのテーブル名"
  }
]
```
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/618752/12b7762f-b548-ea47-c120-2b3842f2eed9.png)



## その他設定
 - **S3 バックアップバケット：**
Firehose用ポリシー記述で、**FirehoseストリームのS3バックアップバケットで指定するS3バケット名** で指定したS3バケットを指定します。
 - **サービスアクセス：**
既存の IAM ロール → 作成したFirehose用ロールを指定します

それ以外は、デフォルト設定のままストリームを作成します。



# テストデータ送信
KDGを使って、作成したFirehoseストリームにテストデータを送信します。
以下記事も参考ください。

https://qiita.com/tamabe/items/c365ace64986f20a9232

以下レコードテンプレートを使って、JSONデータを送信しました。
```json:Firehose送信用JSONレコードテンプレート
{
  "id": {{random.number(1000)}},
  "name": "{{random.arrayElement(["S3Tables", "EMR", "DataFirehose", "Athena", "QuickSight"])}}",
  "value": {{random.number(
    {
      "min": 1,
      "max": 100
    }
  )}}
}
```

※サンプルレコードデータ
`{  "id": 893,  "name": "QuickSight",  "value": 60}`
`{  "id": 684,  "name": "Athena",  "value": 69}`
`{  "id": 107,  "name": "EMR",  "value": 81}`



# (次回記事で詳細)Athenaでクエリ
Athenaで、S3テーブルバケットの該当テーブルをクエリすると、データを確認できました。

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/618752/161e34f2-bd48-0ce2-fe09-f27cf20c1f6a.png)



# 最後に
ここまでで、S3テーブルバケットにデータを送信するところまで確認しました。
次回記事では、このS3テーブルバケットに、AthenaからSQLクエリを実行するまでの設定を確認します。

https://qiita.com/tamabe/private/1c35ced1fee7bad9dc6e


# 参考URL
この記事で参考にしたURLです。
[^13]:AWS記事：Route incoming records to a single Iceberg table
https://docs.aws.amazon.com/firehose/latest/dev/apache-iceberg-format-input-record.html

[^12]:AWS記事：Grant Lake Formation permissions on your table resources
https://docs.aws.amazon.com/AmazonS3/latest/userguide/s3-tables-integrating-aws.html#grant-permissions-tables

[^1]:AWS記事：Amazon S3 table bucket, table, and namespace naming rules
https://docs.aws.amazon.com/AmazonS3/latest/userguide/s3-tables-buckets-naming.html

[^2]:AWS CLI Command Reference：s3tables
https://docs.aws.amazon.com/cli/latest/reference/s3tables/

[^3]:AWS記事：Accessing Amazon S3 tables with Amazon EMR
https://docs.aws.amazon.com/AmazonS3/latest/userguide/s3-tables-integrating-emr.html

[^4]:AWS記事：Querying Amazon S3 tables with Apache Spark
https://docs.aws.amazon.com/AmazonS3/latest/userguide/s3-tables-integrating-open-source-spark.html

[^5]:AWS記事：Streaming data to tables with Amazon Data Firehose
https://docs.aws.amazon.com/AmazonS3/latest/userguide/s3-tables-integrating-firehose.html

[^6]:AWS記事：Register S3 table bucket catalogs in Athena
https://docs.aws.amazon.com/athena/latest/ug/gdc-register-s3-table-bucket-cat.html

[^7]:AWS記事：Visualizing table data with Amazon QuickSight
https://docs.aws.amazon.com/AmazonS3/latest/userguide/s3-tables-integrating-quicksight.html

[^8]:AWS記事：Using Amazon S3 Tables with AWS analytics services #Grant Lake Formation permissions on your table resources
https://docs.aws.amazon.com/AmazonS3/latest/userguide/s3-tables-integrating-aws.html#grant-permissions-tables

[^9]:AWS記事：Using Amazon S3 Tables with AWS analytics services
https://docs.aws.amazon.com/AmazonS3/latest/userguide/s3-tables-integrating-aws.html

[^10]:AWS記事：Use an Iceberg cluster with Spark
https://docs.aws.amazon.com/emr/latest/ReleaseGuide/emr-iceberg-use-spark-cluster.html#emr-iceberg-create-cluster

[^11]:AWS記事：Connect to the Amazon EMR cluster primary node using SSH
https://docs.aws.amazon.com/emr/latest/ManagementGuide/emr-connect-master-node-ssh.html