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

はじめに

本記事は、以下記事の続きです。

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

構成図

最終的な構成図は以下です。

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

実装手順

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

1. S3テーブルバケットを作成
2. EMRからスキーマ作成
3. Firehoseからデータ送信
4. Athenaからクエリ実行
5. QuickSightからデータセット接続

EMRからスキーマ作成手順

以下ステップで行います。

1. EMR用IAMロール ポリシー設定
2. EMR Icebergクラスター作成
3. EMR SparkプライマリノードにSSH接続
4. Sparkセッションを初期化
5. S3テーブルバケットの名前空間 テーブル スキーマ作成
6. S3テーブルバケットを確認
7. EMRリソースのクリーンアップ

EMR用IAMロール ポリシー設定

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

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

作成したEMR用ロールに、AmazonS3TablesFullAccess ポリシーを追加します。

EMR Icebergクラスター作成

Iceberg がインストールされたEMRクラスターを作成¹します。

タグ設定

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

• キー : for-use-with-amazon-emr-managed-policies
• 値 : true

設定ファイル作成

Amazon EMRクラスターで Apache Iceberg を有効にするための設定ファイルを用意します。
※この後のCLIコマンドの--configurations fileで指定します。

configurations.json

[{
    "Classification":"iceberg-defaults",
    "Properties":{"iceberg.enabled":"true"}
}]

キーペア作成

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

コンソールから操作：
EC2コンソール → 「キーペア」タブ → 「キーペアを作成」

ローカルに<キーペア名>.pemファイルが保存されるので、アクセス許可を設定します。

chmod 400 <キーペアファイル格納先パス>

※キーペアファイル格納先パス例：~/.ssh/<キーペア名>.pem

クラスター作成コマンド実行

上記の準備完了後、以下のAWS CLIコマンドを実行してクラスターを作成します。

• デフォルトEMRポリシー(AmazonEMRServicePolicy_v2)を使っている場合は、
  利用VPC, サブネットに指定タグ(for-use-with-amazon-emr-managed-policies=true)を設定していること
• 実行パス配下に設定ファイル(configurations.json)があること
• 実行パス配下にEC2キーペアファイルがあること

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

EMR SparkプライマリノードにSSH接続

今回は、作成したEMRクラスターにSSH接続²を行います。

セキュリティグループにインバウンドルール追加

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

コンソールから操作：
EMRコンソールからクラスターを選択 → プライマリノードのセキュリティグループ

コンソールから操作：
インバウンドのルールを編集 → ルールを追加 → SSH, マイIP → ルールを保存

EMRクラスターにSSH接続

以下コマンドを実行して、SSH接続します。

aws emr ssh --cluster-id <作成したEMRクラスターID> --key-pair-file <キーペアファイル格納先パス> --region ap-northeast-1

問題なくコマンドが通れば、でっかくEMRの文字が出てきます！

Sparkセッションを初期化

SSH接続セッションから、以下コマンドを入力してSparkセッションを初期化します。

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の文字が表示されます。

S3テーブルバケットの名前空間 テーブル スキーマ作成

Sparkシェル上で、S3テーブルバケットを操作するコマンドを実行します。

S3テーブルバケットに名前空間を作成

spark.sql("CREATE NAMESPACE IF NOT EXISTS test.<S3テーブル名前空間名>")

コマンド出力結果例：

次に、S3テーブルバケットにテーブル/スキーマを作成するコマンドを実行します。

S3テーブルバケットにテーブル/スキーマを作成

spark.sql("CREATE TABLE IF NOT EXISTS test.<名前空間名>.`<テーブル名>` ( id INT, name STRING, value INT ) USING iceberg ")

コマンド出力結果例：

S3テーブルバケットを確認

Sparkシェルコマンドで、テーブルのスキーマ定義が想定通り作成されたか確認してみます。

テーブルのスキーマ定義を確認する

spark.sql("DESCRIBE TABLE test.qiita_namespace_tamabe.qiita_table_tamabe").show(false)

コマンド出力結果例：

S3コンソールからも、Sparkシェルで作成した名前空間とテーブルが確認できます。

EMRリソースのクリーンアップ

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

Sparkシェル終了

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

EMRとのSSH接続終了

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

EMRクラスター終了

EMRクラスターを終了

aws emr terminate-clusters --cluster-ids <クラスターID> --region ap-northeast-1

EMRコンソールから終了されていることを確認できます。

最後に

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

参考URL

この記事で参考にしたURLです。

1. 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 ↩

2. 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 ↩

3. AWS公式：EMR よくある質問
   https://aws.amazon.com/jp/emr/faqs/#Billing ↩