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


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

https://qiita.com/tamabe/private/8c459d2434f20191ac3e

今回は、Amazon EMRのApache Icebergクラスターを使用して、
SparkセッションからS3テーブルバケットに接続し、テーブルのスキーマ作成までを行います。
以下AWS公式記事の手順を参考にしています。

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


# 構成図
最終的な構成図は以下です。
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/618752/5f55df06-e41c-e128-d4d1-8891539e154f.png)


本記事では **EMRでS3 Tablesのテーブルとスキーマを作成する** 部分を実装します。


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

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)



# EMRからスキーマ作成手順
以下ステップで行います。
1. [EMR用IAMロール ポリシー設定](#emr用iamロール-ポリシー設定)
1. [EMR Icebergクラスター作成](#emr-icebergクラスター作成)
2. [EMR SparkプライマリノードにSSH接続](#emr-sparkプライマリノードにssh接続)
2. [Sparkセッションを初期化](#sparkセッションを初期化)
2. [S3テーブルバケットの名前空間 テーブル スキーマ作成](#s3テーブルバケットの名前空間-テーブル-スキーマ作成)
2. [S3テーブルバケットを確認](#s3テーブルバケットを確認)
2. [EMRリソースのクリーンアップ](#emrリソースのクリーンアップ)


# EMR用IAMロール ポリシー設定
まず、EMR用のサービスロール/ポリシーを作成します。
コンソール操作：
「IAMコンソール」 → 「ロール」 → 「ロールを作成」

今回、「AWSのサービス」 → 「EMR」を選択して作成します。
このとき、自動で`AmazonEMRServicePolicy_v2`ポリシーがデフォルトで追加されます。

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/618752/e9b3e06d-0c3a-6409-2d8d-dd16dc8adfcc.png)


作成したEMR用ロールに、`AmazonS3TablesFullAccess` ポリシーを追加します。
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/618752/af13ac27-6eb0-93ee-9583-fe5fd560b030.png)




# EMR Icebergクラスター作成
Iceberg がインストールされたEMRクラスターを作成[^10]します。

## タグ設定
デフォルトの`AmazonEMRServicePolicy_v2`ポリシーを利用する場合は、
ポリシー定義内のcondition記述に合わせて、EMRクラスター作成先のVPCとサブネットに
以下の指定タグを設定する必要があるので注意ください。
※タグを設定しないと、この後のCLIコマンドでエラーが出ます。

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/618752/dc59e636-4f1d-83ff-a18f-c8df45dd5521.png)
 - キー : `for-use-with-amazon-emr-managed-policies`
 - 値   : `true`


## 設定ファイル作成
Amazon EMRクラスターで Apache Iceberg を有効にするための設定ファイルを用意します。
※この後のCLIコマンドの`--configurations file`で指定します。
```json:configurations.json
[{
    "Classification":"iceberg-defaults",
    "Properties":{"iceberg.enabled":"true"}
}]
```

## キーペア作成
次に、EC2のキーペアを用意します。
※この後のCLIコマンドの`--ec2-attributes KeyName=`で指定します。

コンソールから操作：
EC2コンソール → 「キーペア」タブ → 「キーペアを作成」
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/618752/362e05c2-87d1-9247-5f6d-a2abf4e2b5f1.png)


ローカルに`<キーペア名>.pem`ファイルが保存されるので、アクセス許可を設定します。
```cli:
chmod 400 <キーペアファイル格納先パス>
```
※キーペアファイル格納先パス例：`~/.ssh/<キーペア名>.pem`


## クラスター作成コマンド実行
上記の準備完了後、以下のAWS CLIコマンドを実行してクラスターを作成します。
 - デフォルトEMRポリシー(`AmazonEMRServicePolicy_v2`)を使っている場合は、
 利用VPC, サブネットに指定タグ(`for-use-with-amazon-emr-managed-policies=true`)を設定していること
 - 実行パス配下に設定ファイル(`configurations.json`)があること
 - 実行パス配下にEC2キーペアファイルがあること

```cli:EMRクラスター作成コマンド
aws emr create-cluster --release-label emr-7.5.0 \
--applications Name=Spark \
--configurations file://configurations.json \
--region ap-northeast-1 \
--name <EMRクラスター名> \
--log-uri s3://<ログファイル保存先S3バケット名>/ \
--instance-type m5.xlarge \
--instance-count 2 \
--service-role <EMRロール名> \
--ec2-attributes KeyName=<キーペアファイル>,InstanceProfile=EMR_EC2_DefaultRole,SubnetId=<サブネットID> \
--tags for-use-with-amazon-emr-managed-policies=true
```

EMRコンソールから、問題なく作成されたかを確認します。
※ステータス「待機中」となっていればOK

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/618752/99693ba2-fb33-086f-6661-5ee77ba985ee.png)




# EMR SparkプライマリノードにSSH接続
今回は、作成したEMRクラスターにSSH接続[^11]を行います。


## セキュリティグループにインバウンドルール追加
EMRクラスターのプライマリノードに関連付けているセキュリティグループに、
SSH(TCPポート22)トラフィックを許可するインバウンドルールを追加します。

コンソールから操作：
EMRコンソールからクラスターを選択 → プライマリノードのセキュリティグループ
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/618752/d792605f-099b-1acb-5741-bf94a004503a.png)


コンソールから操作：
インバウンドのルールを編集 → ルールを追加 → SSH, マイIP → ルールを保存
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/618752/dd8ac0a9-566c-05f9-0e82-4bf41a169d15.png)



## EMRクラスターにSSH接続
以下コマンドを実行して、SSH接続します。
```cli:
aws emr ssh --cluster-id <作成したEMRクラスターID> --key-pair-file <キーペアファイル格納先パス> --region ap-northeast-1
```

問題なくコマンドが通れば、でっかくEMRの文字が出てきます！
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/618752/a97bb0dc-e1c8-f99d-6e44-9a9433fd6a8b.png)




# Sparkセッションを初期化
SSH接続セッションから、以下コマンドを入力してSparkセッションを初期化します。
```cli:Sparkシェル起動コマンド
spark-shell \
--packages software.amazon.s3tables:s3-tables-catalog-for-iceberg-runtime:0.1.3 \
--conf spark.sql.catalog.test=org.apache.iceberg.spark.SparkCatalog \
--conf spark.sql.catalog.test.catalog-impl=software.amazon.s3tables.iceberg.S3TablesCatalog \
--conf spark.sql.catalog.test.warehouse=arn:aws:s3tables:ap-northeast-1:<アカウントID>:bucket/<S3テーブルバケット名> \
--conf spark.sql.extensions=org.apache.iceberg.spark.extensions.IcebergSparkSessionExtensions
```

シェル起動完了したら、Sparkの文字が表示されます。
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/618752/9c2372cb-8c0d-b94f-c7b4-6f5865045375.png)


# S3テーブルバケットの名前空間 テーブル スキーマ作成
Sparkシェル上で、S3テーブルバケットを操作するコマンドを実行します。
```cli:S3テーブルバケットに名前空間を作成
spark.sql("CREATE NAMESPACE IF NOT EXISTS test.<S3テーブル名前空間名>")
```

コマンド出力結果例：
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/618752/b532560c-6de8-5c73-0314-63ce2744e0bd.png)


次に、S3テーブルバケットにテーブル/スキーマを作成するコマンドを実行します。
```cli:S3テーブルバケットにテーブル/スキーマを作成
spark.sql("CREATE TABLE IF NOT EXISTS test.<名前空間名>.`<テーブル名>` ( id INT, name STRING, value INT ) USING iceberg ")
```

コマンド出力結果例：
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/618752/14cb138e-ae9d-0743-097e-bd3e301385d4.png)


# S3テーブルバケットを確認
Sparkシェルコマンドで、テーブルのスキーマ定義が想定通り作成されたか確認してみます。
```cli:テーブルのスキーマ定義を確認する
spark.sql("DESCRIBE TABLE test.qiita_namespace_tamabe.qiita_table_tamabe").show(false)
```

コマンド出力結果例：
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/618752/f8f60049-9978-2b98-0251-39d2d0af5c24.png)


S3コンソールからも、Sparkシェルで作成した名前空間とテーブルが確認できます。
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/618752/cdcc49b4-0e6e-8325-8628-53b971f0de85.png)



# EMRリソースのクリーンアップ
テーブル/スキーマの作成が終わったので、不要となったEMRクラスターを終了します。
※EMRはシャットダウンするまで利用時間あたりの課金が発生します[^12]。

## Sparkシェル終了
Sparkシェルは、`Ctrl + C`で終了できます。

## EMRとのSSH接続終了
EMRとのSSH接続は、`exit`コマンドで終了します。

## EMRクラスター終了
```cli:EMRクラスターを終了
aws emr terminate-clusters --cluster-ids <クラスターID> --region ap-northeast-1
```
EMRコンソールから終了されていることを確認できます。
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/618752/b9eb215f-5c0c-3416-7dfb-86d486eb8635.png)



# 最後に
ここまでで、S3テーブルバケットにテーブル/スキーマを作成しました。
次回記事では、このS3テーブルバケットに、Data Firehoseからデータを送信してみます。

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


# 参考URL
この記事で参考にしたURLです。
[^12]:AWS公式：EMR よくある質問
https://aws.amazon.com/jp/emr/faqs/#Billing

[^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