0
0

Delete article

Deleted articles cannot be recovered.

Draft of this article would be also deleted.

Are you sure you want to delete this article?

@naritomo08(nari tomo)

【第7回】EMR ServerlessとEventBridge SchedulerでIcebergメンテナンスを定期実行する

0
Last updated at Posted at 2026-09-20

はじめに

別記事では、S3 rawを7日、batch Icebergを14日、Flink realtime Icebergを当日分だけ保持する手順を作成した。本記事では、そのIceberg maintenance用PySpark jobをAmazon EMR Serverlessへ配置し、Amazon EventBridge Schedulerから毎日自動実行する。

awslog-emr-iceberg-maintenance.png

S3 rawの7日削除はS3 Lifecycleが自動実行するため、本記事のschedule対象には含めない。

前提

本記事は単独で最初に実施するものではない。次の順序で構築した環境を前提とする。

  1. 「HDFSログ分析基盤をAWS S3へ段階移行する」シリーズの第1〜7編を完了する
  2. 「オンプレのログ分析基盤をAWSマネージドサービスへ移行する」シリーズの第1〜6回を完了する
  3. 「Amazon S3のrawログとApache Icebergに保持期間とメンテナンスを設定する」を実施し、保持期間とPySpark jobを確認する
  4. 本記事でPySpark jobをAWSから定期実行する

必須となる記事

記事 本記事で引き継ぐもの
【第3編】Apache OzoneのローカルS3からAWS S3へrawログとIcebergを移行してみる S3 Iceberg、Glue Database/Table
【第3回】Spark batchをEMR Serverlessへ移行する EMR Serverless application、artifact bucket、Spark/Iceberg設定
【第4回】バックアップ済みApache IcebergとAWS Glue Data Catalogを引き継ぐ 既存table名、location、metadata location
【第6回】FlinkをManaged Service for Apache Flinkへ移行する Flink realtime tableとcheckpoint運用
Amazon S3のrawログとApache Icebergに保持期間とメンテナンスを設定する 保持期間、maintenance用PySpark file、手動検証方法

既存S3 backupを利用してS3移行シリーズ第2〜3編を省略した場合でも、S3 IcebergとGlue Tableが構築済みで、Athenaから参照できればよい。

開始条件

  • EMR ServerlessのSpark applicationを作成済み
  • artifact用S3 bucketを作成済み
  • S3 IcebergとAWS Glue Data Catalogのlogs databaseを構築済み
  • batch用のsyslog_iceberg、authlog_icebergが存在する
  • realtime用のsyslog_events、authlog_events、syslog_host_1m、authlog_host_1mが存在する
  • maintenance用PySpark fileを手動実行し、削除対象とorphan fileのdry-runを確認済み
  • Managed Service for Apache FlinkのcheckpointとIceberg commitが正常である
  • Lake Formationを使用する場合、runtime roleへSELECT、INSERT、DELETE、ALTER相当の権限を付与できる

6 tableのいずれかが存在しない場合は、本記事のscheduleを先に作らず、tableの構築またはtable listの修正を行う。

値を設定する。

export AWS_REGION=ap-northeast-1
export AWS_PROFILE=home-aws-admin

source /etc/iceberg/aws.env
export LAKE_BUCKET="${ICEBERG_BUCKET}"
export GLUE_DATABASE="${GLUE_DATABASE:-logs}"

export EMR_APPLICATION_ID="${EMR_APPLICATION_ID:?set EMR Serverless application ID}"
export ARTIFACT_BUCKET="${ARTIFACT_BUCKET:?set artifact bucket name}"
export MAINTENANCE_SCRIPT=iceberg_retention_maintenance.py
export MAINTENANCE_SCRIPT_KEY=jobs/iceberg_retention_maintenance.py
export MAINTENANCE_JOB_ROLE=home-iceberg-maintenance-job-role
export MAINTENANCE_SCHEDULER_ROLE=home-iceberg-maintenance-scheduler-role
export MAINTENANCE_SCHEDULE=home-iceberg-maintenance-daily

export AWS_ACCOUNT_ID=$(aws sts get-caller-identity \
  --query Account --output text --profile "${AWS_PROFILE}")

aws emr-serverless get-application \
  --region "${AWS_REGION}" \
  --application-id "${EMR_APPLICATION_ID}" \
  --profile "${AWS_PROFILE}"

aws glue get-tables \
  --region "${AWS_REGION}" \
  --database-name "${GLUE_DATABASE}" \
  --query 'TableList[].Name' \
  --profile "${AWS_PROFILE}"

maintenance jobをS3へ配置する

独立したmaintenance記事で作成したiceberg_retention_maintenance.pyをuploadする。

aws s3 cp "${MAINTENANCE_SCRIPT}" \
  "s3://${ARTIFACT_BUCKET}/${MAINTENANCE_SCRIPT_KEY}" \
  --region "${AWS_REGION}" \
  --profile "${AWS_PROFILE}"

aws s3api head-object \
  --bucket "${ARTIFACT_BUCKET}" \
  --key "${MAINTENANCE_SCRIPT_KEY}" \
  --profile "${AWS_PROFILE}"

EMR Serverless runtime roleを作成する

trust policyをemr-job-trust.jsonとして保存する。

{
  "Version": "2012-10-17",
  "Statement": [{
    "Effect": "Allow",
    "Principal": {"Service": "emr-serverless.amazonaws.com"},
    "Action": "sts:AssumeRole"
  }]
}

aws iam create-role \
  --role-name "${MAINTENANCE_JOB_ROLE}" \
  --assume-role-policy-document file://emr-job-trust.json \
  --profile "${AWS_PROFILE}"

runtime policyをemr-maintenance-policy.jsonとして保存する。Iceberg procedureはdata fileを読み書き・削除するため、warehouse prefixにはDeleteObjectも必要になる。

{
  "Version": "2012-10-17",
  "Statement": [
    {
      "Sid": "ListRequiredBuckets",
      "Effect": "Allow",
      "Action": "s3:ListBucket",
      "Resource": [
        "arn:aws:s3:::YOUR_LAKE_BUCKET",
        "arn:aws:s3:::YOUR_ARTIFACT_BUCKET"
      ]
    },
    {
      "Sid": "ReadMaintenanceScript",
      "Effect": "Allow",
      "Action": "s3:GetObject",
      "Resource": "arn:aws:s3:::YOUR_ARTIFACT_BUCKET/jobs/iceberg_retention_maintenance.py"
    },
    {
      "Sid": "MaintainIcebergWarehouse",
      "Effect": "Allow",
      "Action": ["s3:GetObject", "s3:PutObject", "s3:DeleteObject", "s3:AbortMultipartUpload"],
      "Resource": "arn:aws:s3:::YOUR_LAKE_BUCKET/warehouse/*"
    },
    {
      "Sid": "GlueCatalog",
      "Effect": "Allow",
      "Action": [
        "glue:GetDatabase", "glue:GetDatabases",
        "glue:GetTable", "glue:GetTables",
        "glue:GetPartition", "glue:GetPartitions",
        "glue:UpdateTable"
      ],
      "Resource": "*"
    }
  ]
}

placeholderを置換してinline policyを登録する。

sed \
  -e "s/YOUR_LAKE_BUCKET/${LAKE_BUCKET}/g" \
  -e "s/YOUR_ARTIFACT_BUCKET/${ARTIFACT_BUCKET}/g" \
  emr-maintenance-policy.json > emr-maintenance-policy.rendered.json

aws iam put-role-policy \
  --role-name "${MAINTENANCE_JOB_ROLE}" \
  --policy-name IcebergMaintenanceAccess \
  --policy-document file://emr-maintenance-policy.rendered.json \
  --profile "${AWS_PROFILE}"

export MAINTENANCE_JOB_ROLE_ARN="arn:aws:iam::${AWS_ACCOUNT_ID}:role/${MAINTENANCE_JOB_ROLE}"

SSE-KMSを使用している場合は対象KMS keyのDecrypt、GenerateDataKeyを追加する。Lake Formationでtable locationを管理している場合は、IAMに加えてdata lake permissionも付与する。

手動実行で確認する

schedule作成前に1回だけ手動実行する。

SPARK_PARAMETERS="--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://${LAKE_BUCKET}/warehouse --conf spark.sql.session.timeZone=Asia/Tokyo"

MANUAL_JOB_RUN_ID=$(aws emr-serverless start-job-run \
  --region "${AWS_REGION}" \
  --application-id "${EMR_APPLICATION_ID}" \
  --execution-role-arn "${MAINTENANCE_JOB_ROLE_ARN}" \
  --name iceberg-maintenance-manual \
  --job-driver "$(jq -nc \
    --arg entry "s3://${ARTIFACT_BUCKET}/${MAINTENANCE_SCRIPT_KEY}" \
    --arg params "${SPARK_PARAMETERS}" \
    '{sparkSubmit:{entryPoint:$entry,sparkSubmitParameters:$params}}')" \
  --execution-timeout-minutes 180 \
  --query jobRunId --output text \
  --profile "${AWS_PROFILE}")

aws emr-serverless get-job-run \
  --region "${AWS_REGION}" \
  --application-id "${EMR_APPLICATION_ID}" \
  --job-run-id "${MANUAL_JOB_RUN_ID}" \
  --profile "${AWS_PROFILE}"

jobがSUCCESSになり、6 tableの保持期間、snapshot、Flink checkpointが正常であることを確認してから定期実行へ進む。

EventBridge Scheduler実行roleを作成する

trust policyをscheduler-trust.jsonとして保存する。

{
  "Version": "2012-10-17",
  "Statement": [{
    "Effect": "Allow",
    "Principal": {"Service": "scheduler.amazonaws.com"},
    "Action": "sts:AssumeRole"
  }]
}

aws iam create-role \
  --role-name "${MAINTENANCE_SCHEDULER_ROLE}" \
  --assume-role-policy-document file://scheduler-trust.json \
  --profile "${AWS_PROFILE}"

export MAINTENANCE_SCHEDULER_ROLE_ARN="arn:aws:iam::${AWS_ACCOUNT_ID}:role/${MAINTENANCE_SCHEDULER_ROLE}"
export EMR_APPLICATION_ARN="arn:aws:emr-serverless:${AWS_REGION}:${AWS_ACCOUNT_ID}:/applications/${EMR_APPLICATION_ID}"

Schedulerには対象applicationのStartJobRunと、maintenance runtime roleだけを渡せるPassRoleを許可する。scheduler-policy.jsonとして保存する。

{
  "Version": "2012-10-17",
  "Statement": [
    {
      "Effect": "Allow",
      "Action": "emr-serverless:StartJobRun",
      "Resource": "EMR_APPLICATION_ARN"
    },
    {
      "Effect": "Allow",
      "Action": "iam:PassRole",
      "Resource": "MAINTENANCE_JOB_ROLE_ARN",
      "Condition": {
        "StringEquals": {
          "iam:PassedToService": "emr-serverless.amazonaws.com"
        }
      }
    }
  ]
}

sed \
  -e "s|EMR_APPLICATION_ARN|${EMR_APPLICATION_ARN}|g" \
  -e "s|MAINTENANCE_JOB_ROLE_ARN|${MAINTENANCE_JOB_ROLE_ARN}|g" \
  scheduler-policy.json > scheduler-policy.rendered.json

aws iam put-role-policy \
  --role-name "${MAINTENANCE_SCHEDULER_ROLE}" \
  --policy-name StartIcebergMaintenance \
  --policy-document file://scheduler-policy.rendered.json \
  --profile "${AWS_PROFILE}"

毎日01:30に定期実行する

EventBridge Schedulerのuniversal targetからEMR ServerlessのStartJobRun APIを呼ぶ。timezoneを指定するため、夏時間やUTC換算をcommandへ埋め込まない。

TARGET_INPUT=$(jq -nc \
  --arg applicationId "${EMR_APPLICATION_ID}" \
  --arg executionRoleArn "${MAINTENANCE_JOB_ROLE_ARN}" \
  --arg entryPoint "s3://${ARTIFACT_BUCKET}/${MAINTENANCE_SCRIPT_KEY}" \
  --arg params "${SPARK_PARAMETERS}" \
  '{
    clientToken:"<aws.scheduler.execution-id>",
    applicationId:$applicationId,
    executionRoleArn:$executionRoleArn,
    name:"iceberg-maintenance-scheduled",
    executionTimeoutMinutes:180,
    jobDriver:{
      sparkSubmit:{
        entryPoint:$entryPoint,
        sparkSubmitParameters:$params
      }
    }
  }')

TARGET=$(jq -nc \
  --arg roleArn "${MAINTENANCE_SCHEDULER_ROLE_ARN}" \
  --arg input "${TARGET_INPUT}" \
  '{
    RoleArn:$roleArn,
    Arn:"arn:aws:scheduler:::aws-sdk:emr-serverless:startJobRun",
    Input:$input,
    RetryPolicy:{MaximumEventAgeInSeconds:3600,MaximumRetryAttempts:0}
  }')

aws scheduler create-schedule \
  --region "${AWS_REGION}" \
  --name "${MAINTENANCE_SCHEDULE}" \
  --schedule-expression 'cron(30 1 * * ? *)' \
  --schedule-expression-timezone Asia/Tokyo \
  --flexible-time-window '{"Mode":"OFF"}' \
  --target "${TARGET}" \
  --state ENABLED \
  --description 'Daily Iceberg retention and maintenance' \
  --profile "${AWS_PROFILE}"

maintenanceは再実行で追加削除が発生し得るため、Schedulerの自動retryは無効にしている。job timeoutは3時間に制限し、前日のjobが終了していない場合はscheduleを一時停止して原因を調査する。

定期実行を確認する

aws scheduler get-schedule \
  --region "${AWS_REGION}" \
  --name "${MAINTENANCE_SCHEDULE}" \
  --profile "${AWS_PROFILE}"

aws emr-serverless list-job-runs \
  --region "${AWS_REGION}" \
  --application-id "${EMR_APPLICATION_ID}" \
  --max-results 10 \
  --profile "${AWS_PROFILE}"

最低限、次を監視する。

  • scheduleの次回実行時刻と状態
  • EMR Serverless jobのSUCCESS／FAILED
  • job durationと費用
  • table別のMIN(dt)、MAX(dt)、件数
  • snapshot数、data file数、平均file size
  • Flink checkpoint成功時刻とconsumer lag

一時停止と再開

Flinkの復旧中やtable障害の調査中は、先にscheduleを無効化する。

aws scheduler update-schedule \
  --region "${AWS_REGION}" \
  --name "${MAINTENANCE_SCHEDULE}" \
  --schedule-expression 'cron(30 1 * * ? *)' \
  --schedule-expression-timezone Asia/Tokyo \
  --flexible-time-window '{"Mode":"OFF"}' \
  --target "${TARGET}" \
  --state DISABLED \
  --profile "${AWS_PROFILE}"

確認後、同じcommandの--stateをENABLEDへ戻す。

本記事で作成したresourceを削除する

S3 raw、Iceberg table、Glue Database、EMR Serverless applicationは継続利用するため削除しない。本記事で追加したschedule、IAM role、job fileだけを逆順で削除する。

1. scheduleを無効化して削除する

実行中jobがないことを確認してから削除する。

aws emr-serverless list-job-runs \
  --region "${AWS_REGION}" \
  --application-id "${EMR_APPLICATION_ID}" \
  --states SUBMITTED PENDING SCHEDULED RUNNING \
  --profile "${AWS_PROFILE}"

aws scheduler delete-schedule \
  --region "${AWS_REGION}" \
  --name "${MAINTENANCE_SCHEDULE}" \
  --profile "${AWS_PROFILE}"

2. Scheduler roleを削除する

aws iam delete-role-policy \
  --role-name "${MAINTENANCE_SCHEDULER_ROLE}" \
  --policy-name StartIcebergMaintenance \
  --profile "${AWS_PROFILE}"

aws iam delete-role \
  --role-name "${MAINTENANCE_SCHEDULER_ROLE}" \
  --profile "${AWS_PROFILE}"

3. maintenance runtime roleを削除する

aws iam delete-role-policy \
  --role-name "${MAINTENANCE_JOB_ROLE}" \
  --policy-name IcebergMaintenanceAccess \
  --profile "${AWS_PROFILE}"

aws iam delete-role \
  --role-name "${MAINTENANCE_JOB_ROLE}" \
  --profile "${AWS_PROFILE}"

4. maintenance job fileを削除する

削除対象を固定してから、この記事でuploadしたobjectだけを削除する。

aws s3api head-object \
  --bucket "${ARTIFACT_BUCKET}" \
  --key "${MAINTENANCE_SCRIPT_KEY}" \
  --profile "${AWS_PROFILE}"

aws s3api delete-object \
  --bucket "${ARTIFACT_BUCKET}" \
  --key "${MAINTENANCE_SCRIPT_KEY}" \
  --profile "${AWS_PROFILE}"

versioningが有効なartifact bucketでは、この操作はdelete markerを追加する。完全削除が必要な場合だけ、対象keyのversionを確認して削除する。

5. 削除結果を確認する

aws scheduler get-schedule \
  --region "${AWS_REGION}" \
  --name "${MAINTENANCE_SCHEDULE}" \
  --profile "${AWS_PROFILE}" || true

aws iam get-role --role-name "${MAINTENANCE_SCHEDULER_ROLE}" \
  --profile "${AWS_PROFILE}" || true
aws iam get-role --role-name "${MAINTENANCE_JOB_ROLE}" \
  --profile "${AWS_PROFILE}" || true

aws s3api head-object \
  --bucket "${ARTIFACT_BUCKET}" \
  --key "${MAINTENANCE_SCRIPT_KEY}" \
  --profile "${AWS_PROFILE}" || true

参考資料

関連記事

実機検証について

本手順は執筆時点では一連の実機検証を完了していない。universal targetのservice identifier、EMR release、Iceberg procedure、Lake Formation permissionを実施時の環境で確認し、最初は手動jobを成功させてからscheduleを有効化する。

0
0
0

Register as a new user and use Qiita more conveniently

  1. You get articles that match your needs
  2. You can efficiently read back useful information
  3. You can use dark theme
What you can do with signing up
Sign upLogin
0
0

Delete article

Deleted articles cannot be recovered.

Draft of this article would be also deleted.

Are you sure you want to delete this article?