はじめに
先行シリーズで、batch tableとstreaming tableはAmazon S3へ移行され、AWS Glue Data Catalogのglue_prod.logsから参照できる。本稿ではdatabaseやtableを新規作成せず、その既存IcebergをEMR ServerlessとManaged Service for Apache Flinkから安全に引き継ぐ。
1. 既存環境とprofileを読み込む
source /etc/iceberg/aws.env
export AWS_PROFILE=home-aws
aws sts get-caller-identity --profile "${AWS_PROFILE}"
aws glue get-database \
--region "${AWS_REGION}" \
--name "${GLUE_DATABASE}" \
--profile "${AWS_PROFILE}"
2. 既存batch tableを確認する
既存のsyslog_icebergとauthlog_icebergに対してCREATE TABLEや同じlocationを指定したDDLを再実行しない。Glueのtable名、Iceberg metadata location、S3 bucketが既存環境ファイルと一致することを確認する。
aws glue get-tables \
--region "${AWS_REGION}" \
--database-name "${GLUE_DATABASE}" \
--query 'TableList[].{Name:Name,Type:TableType,Metadata:Parameters.metadata_location}' \
--profile "${AWS_PROFILE}"
aws s3 ls \
"s3://${ICEBERG_BUCKET}/warehouse/${GLUE_DATABASE}/" \
--profile "${AWS_PROFILE}"
3. 既存streaming tableを確認する
syslog_events、authlog_events、syslog_host_1m、authlog_host_1mは先行シリーズで作成済みであるため、通常は以下のDDLとtable作成jobをスキップする。既存tableがない場合に限り、既存batch tableとは異なるlocationへ作成する。
CREATE TABLE IF NOT EXISTS glue_prod.logs.syslog_events (
ts_raw string, ts timestamp_ntz, host string, program string,
msg string, dt date, hr int
) USING iceberg
LOCATION 's3://${ICEBERG_BUCKET}/warehouse/${GLUE_DATABASE}/syslog_events'
PARTITIONED BY (dt, host)
TBLPROPERTIES ('format-version'='2', 'write.distribution-mode'='hash');
CREATE TABLE IF NOT EXISTS glue_prod.logs.authlog_events (
ts_raw string, ts timestamp_ntz, host string, program string,
msg string, dt date, hr int
) USING iceberg
LOCATION 's3://${ICEBERG_BUCKET}/warehouse/${GLUE_DATABASE}/authlog_events'
PARTITIONED BY (dt, host)
TBLPROPERTIES ('format-version'='2', 'write.distribution-mode'='hash');
CREATE TABLE IF NOT EXISTS glue_prod.logs.syslog_host_1m (
window_start timestamp_ntz, host string, cnt bigint, dt date, hr int
) USING iceberg
LOCATION 's3://${ICEBERG_BUCKET}/warehouse/${GLUE_DATABASE}/syslog_host_1m'
PARTITIONED BY (dt)
TBLPROPERTIES ('format-version'='2');
CREATE TABLE IF NOT EXISTS glue_prod.logs.authlog_host_1m (
window_start timestamp_ntz, host string, cnt bigint, dt date, hr int
) USING iceberg
LOCATION 's3://${ICEBERG_BUCKET}/warehouse/${GLUE_DATABASE}/authlog_host_1m'
PARTITIONED BY (dt)
TBLPROPERTIES ('format-version'='2');
上のDDLを実行するPySpark fileを用意し、第3回のEMR Serverless applicationから一度だけ実行する。
# create_streaming_tables.py
import sys
from pyspark.sql import SparkSession
bucket, database = sys.argv[1], sys.argv[2]
spark = SparkSession.builder.appName("create-streaming-tables").getOrCreate()
event_columns = """
ts_raw string, ts timestamp_ntz, host string, program string,
msg string, dt date, hr int
"""
minute_columns = """
window_start timestamp_ntz, host string, cnt bigint, dt date, hr int
"""
for table in ("syslog_events", "authlog_events"):
spark.sql(f"""
CREATE TABLE IF NOT EXISTS glue_prod.{database}.{table} ({event_columns})
USING iceberg
LOCATION 's3://{bucket}/warehouse/{database}/{table}'
PARTITIONED BY (dt, host)
TBLPROPERTIES ('format-version'='2', 'write.distribution-mode'='hash')
""")
for table in ("syslog_host_1m", "authlog_host_1m"):
spark.sql(f"""
CREATE TABLE IF NOT EXISTS glue_prod.{database}.{table} ({minute_columns})
USING iceberg
LOCATION 's3://{bucket}/warehouse/{database}/{table}'
PARTITIONED BY (dt)
TBLPROPERTIES ('format-version'='2')
""")
aws s3 cp create_streaming_tables.py \
"s3://${ARTIFACT_BUCKET}/jobs/create_streaming_tables.py" \
--region "${AWS_REGION}" --profile home-aws-admin
DDL_JOB_RUN_ID=$(aws emr-serverless start-job-run \
--region "${AWS_REGION}" \
--application-id "${EMR_APPLICATION_ID}" \
--execution-role-arn "${EMR_JOB_ROLE_ARN}" \
--job-driver "{\"sparkSubmit\":{\"entryPoint\":\"s3://${ARTIFACT_BUCKET}/jobs/create_streaming_tables.py\",\"entryPointArguments\":[\"${ICEBERG_BUCKET}\",\"${GLUE_DATABASE}\"],\"sparkSubmitParameters\":\"--conf spark.sql.extensions=org.apache.iceberg.spark.extensions.IcebergSparkSessionExtensions --conf spark.sql.catalog.glue_prod=org.apache.iceberg.spark.SparkCatalog --conf spark.sql.catalog.glue_prod.catalog-impl=org.apache.iceberg.aws.glue.GlueCatalog --conf spark.sql.catalog.glue_prod.io-impl=org.apache.iceberg.aws.s3.S3FileIO --conf spark.sql.catalog.glue_prod.warehouse=s3://${ICEBERG_BUCKET}/warehouse\"}}" \
--query jobRunId --output text --profile home-aws-admin)
while :; do
DDL_JOB_STATE=$(aws emr-serverless get-job-run \
--region "${AWS_REGION}" --application-id "${EMR_APPLICATION_ID}" \
--job-run-id "${DDL_JOB_RUN_ID}" --query jobRun.state --output text \
--profile home-aws-admin)
case "${DDL_JOB_STATE}" in
SUCCESS) break ;;
FAILED|CANCELLING|CANCELLED) exit 1 ;;
esac
sleep 30
done
create_streaming_tables.pyではSparkSessionを作成し、上記4つのCREATE TABLE IF NOT EXISTSをspark.sql()へ渡す。bucketとdatabaseはsys.argv[1]、sys.argv[2]から受け取り、DDLへ固定値を書かない。
4. Glueを再確認する
aws glue get-tables --region "$AWS_REGION" --database-name "$GLUE_DATABASE" \
--query 'TableList[].{Name:Name,Type:TableType,Metadata:Parameters.metadata_location}' \
--profile "${AWS_PROFILE}"
metadata_locationが各tableのmetadata/*.metadata.jsonを指すことを確認する。S3上にdata fileだけがありGlue entryがない状態、またはGlueが古いmetadataを指す状態は完了ではない。
migration方針
Icebergのmetadata directoryを再コピーしない。過去dataと既存tableをそのまま使い、EMR ServerlessとManaged Service for Apache Flinkは切り替え後のdataを既存tableへ書く。既存tableのmetadata_locationを手作業で変更しない。
権限
writer roleにはGlue database/tableのread/writeと対象S3 prefixのread/writeを与える。home-aws profileや/var/lib/spark/.aws/credentialsをAWSサービスへ配布しない。Athena利用者は原則read-onlyにする。Lake Formationを有効にしたaccountではIAMだけでなくData Lake permissionも確認する。
schema互換性の確認
DESCRIBE glue_prod.logs.syslog_iceberg;
DESCRIBE glue_prod.logs.authlog_iceberg;
SELECT * FROM glue_prod.logs.syslog_iceberg.snapshots ORDER BY committed_at DESC;
column名、型、順序、partition spec、format-version、locationをオンプレ側と比較する。timestampとtimestamp_ntzの差は時刻ずれにつながるため、sample 5件を文字列化して比較する。
参考資料
関連記事
- 【導入編】オンプレのログ分析基盤をAWSマネージドサービスへ移行する全体設計
- 【第3回】S3 Rawログの日次Spark処理をEMR Serverlessへ移行する
- 【第5回】TrinoのIceberg分析をAmazon Athenaへ移行する
実機検証について
本稿は未検証である。Lake Formationの既定設定はaccountごとに異なるため、permission modelを確認する。