はじめに
本記事は、以下記事の続きです。
今回は、Amazon EMRのApache Icebergクラスターを使用して、
SparkセッションからS3テーブルバケットに接続し、テーブルのスキーマ作成までを行います。
以下AWS公式記事の手順を参考にしています。
構成図
最終的な構成図は以下です。
本記事では EMRでS3 Tablesのテーブルとスキーマを作成する 部分を実装します。
実装手順
以下ステップで実装していきます。
ステップ3～5では、権限管理でLake Formationも出てきます。
本記事では 2. EMRからスキーマ作成 のステップを行っていきます。
EMRからスキーマ作成手順
以下ステップで行います。
- EMR用IAMロール ポリシー設定
- EMR Icebergクラスター作成
- EMR SparkプライマリノードにSSH接続
- Sparkセッションを初期化
- S3テーブルバケットの名前空間 テーブル スキーマ作成
- S3テーブルバケットを確認
- EMRリソースのクリーンアップ
EMR用IAMロール ポリシー設定
まず、EMR用のサービスロール/ポリシーを作成します。
コンソール操作：
「IAMコンソール」 → 「ロール」 → 「ロールを作成」
今回、「AWSのサービス」 → 「EMR」を選択して作成します。
このとき、自動でAmazonEMRServicePolicy_v2ポリシーがデフォルトで追加されます。
作成したEMR用ロールに、AmazonS3TablesFullAccess ポリシーを追加します。
EMR Icebergクラスター作成
Iceberg がインストールされたEMRクラスターを作成1します。
タグ設定
デフォルトのAmazonEMRServicePolicy_v2ポリシーを利用する場合は、
ポリシー定義内のcondition記述に合わせて、EMRクラスター作成先のVPCとサブネットに
以下の指定タグを設定する必要があるので注意ください。
※タグを設定しないと、この後のCLIコマンドでエラーが出ます。
- キー :
for-use-with-amazon-emr-managed-policies - 値 :
true
設定ファイル作成
Amazon EMRクラスターで Apache Iceberg を有効にするための設定ファイルを用意します。
※この後のCLIコマンドの--configurations fileで指定します。
[{
"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キーペアファイルがあること
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接続2を行います。
セキュリティグループにインバウンドルール追加
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-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テーブルバケットを操作するコマンドを実行します。
spark.sql("CREATE NAMESPACE IF NOT EXISTS test.<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はシャットダウンするまで利用時間あたりの課金が発生します3。
Sparkシェル終了
Sparkシェルは、Ctrl + Cで終了できます。
EMRとのSSH接続終了
EMRとのSSH接続は、exitコマンドで終了します。
EMRクラスター終了
aws emr terminate-clusters --cluster-ids <クラスターID> --region ap-northeast-1
EMRコンソールから終了されていることを確認できます。
最後に
ここまでで、S3テーブルバケットにテーブル/スキーマを作成しました。
次回記事では、このS3テーブルバケットに、Data Firehoseからデータを送信してみます。
参考URL
この記事で参考にしたURLです。
-
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 ↩ -
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 ↩ -
AWS公式：EMR よくある質問
https://aws.amazon.com/jp/emr/faqs/#Billing ↩