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)

【第3回】S3 Rawログの日次Spark処理をEMR Serverlessへ移行する

0
Last updated at Posted at 2026-09-19

はじめに

オンプレのSpark 3.5系日次batchをEMR Serverlessへ移す。初回検証では、入力に先行シリーズから継続利用しているS3 raw、比較先に既存のS3 Iceberg tableを使う。検証が一致してから、MSK Connect切り替え後の日付だけを既存Iceberg v2 tableへ追加する。

source /etc/rawlog-s3-backup/common.env
export RAW_BUCKET="${S3_BUCKET_NAME}"
export AWS_RAW_PROFILE=home-rawlog-backup

source /etc/iceberg/aws.env
export LAKE_BUCKET="${ICEBERG_BUCKET}"
export AWS_SYNC_PROFILE=home-aws

aws sts get-caller-identity --profile "${AWS_RAW_PROFILE}"
aws sts get-caller-identity --profile "${AWS_SYNC_PROFILE}"

これらのprofileは手元で入力と結果を確認するために使う。EMR Serverless jobにはcredential fileを渡さず、既存Raw prefixのread権限と既存Iceberg/Glueへのread/write権限をruntime roleへ付与する。

schemaを維持する

host string, ts timestamp_ntz, severity int,
program string, msg string, dt date, hr int
PARTITIONED BY (dt, host)

raw JSONの実フィールドはts、src_host、program、messageである。severityは既存変換規則を移植し、値を想像で変更しない。

1. releaseを選ぶ

オンプレSpark 3.5系と揃うEMR 7系releaseを選ぶ。執筆時の公式release historyではEMR 7.13.0がSpark 3.5.6を含むが、利用regionのEMR Serverlessで選択可能かを実行時に確認する。

export AWS_ADMIN_PROFILE=home-aws-admin
export EMR_RELEASE_LABEL=emr-7.13.0

EMR_APPLICATION_ID=$(aws emr-serverless create-application \
  --region "$AWS_REGION" \
  --name log-batch \
  --release-label "${EMR_RELEASE_LABEL}" \
  --type SPARK \
  --auto-start-configuration enabled=true \
  --auto-stop-configuration enabled=true,idleTimeoutMinutes=15 \
  --query applicationId --output text \
  --profile "${AWS_ADMIN_PROFILE}")

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

2. runtime roleを作る

roleには次を付与する。

  • raw bucketのListBucket、GetObject
  • lake bucketのListBucket、GetObject、PutObject、DeleteObject
  • script/log prefixへのaccess
  • 既存Glue database/tableのGet*、UpdateTable
  • SSE-KMS利用時のKMS権限

trust principalはemr-serverless.amazonaws.comとする。

cat > /tmp/emr-serverless-trust.json <<'EOF'
{"Version":"2012-10-17","Statement":[{"Effect":"Allow","Principal":{"Service":"emr-serverless.amazonaws.com"},"Action":"sts:AssumeRole"}]}
EOF

EMR_JOB_ROLE_ARN=$(aws iam create-role \
  --role-name home-log-emr-serverless-role \
  --assume-role-policy-document file:///tmp/emr-serverless-trust.json \
  --query Role.Arn --output text --profile "${AWS_ADMIN_PROFILE}")

jq -n --arg raw "${RAW_BUCKET}" --arg lake "${LAKE_BUCKET}" \
  --arg artifacts "${ARTIFACT_BUCKET}" '{Version:"2012-10-17",Statement:[
    {Effect:"Allow",Action:"s3:ListBucket",Resource:[("arn:aws:s3:::"+$raw),("arn:aws:s3:::"+$lake),("arn:aws:s3:::"+$artifacts)]},
    {Effect:"Allow",Action:"s3:GetObject",Resource:[("arn:aws:s3:::"+$raw+"/raw/*"),("arn:aws:s3:::"+$artifacts+"/jobs/*"),("arn:aws:s3:::"+$lake+"/warehouse/*")]},
    {Effect:"Allow",Action:["s3:PutObject","s3:DeleteObject"],Resource:("arn:aws:s3:::"+$lake+"/warehouse/*")},
    {Effect:"Allow",Action:["glue:GetDatabase","glue:GetDatabases","glue:GetTable","glue:GetTables","glue:UpdateTable"],Resource:"*"}
  ]}' > /tmp/emr-serverless-policy.json

aws iam put-role-policy \
  --role-name home-log-emr-serverless-role \
  --policy-name home-log-emr-serverless-policy \
  --policy-document file:///tmp/emr-serverless-policy.json \
  --profile "${AWS_ADMIN_PROFILE}"

SSE-KMSを使用するbucketでは、対象keyのkms:Decryptとkms:GenerateDataKeyもinline policyへ追加する。動作確認後はGlueのResourceをcatalog、database、table ARNへ絞る。

3. PySpark jobを作る

from datetime import date, timedelta
import sys
from pyspark.sql import SparkSession, functions as F, types as T

target = sys.argv[1] if len(sys.argv) > 1 else str(date.today() - timedelta(days=1))
mode = sys.argv[2] if len(sys.argv) > 2 else "validate"
raw_bucket = sys.argv[3]
if mode not in ("validate", "append"):
    raise ValueError("mode must be validate or append")

spark = (SparkSession.builder
    .appName("raw-to-iceberg")
    .config("spark.sql.session.timeZone", "Asia/Tokyo")
    .getOrCreate())

raw_schema = T.StructType([
    T.StructField("ts", T.StringType()),
    T.StructField("src_host", T.StringType()),
    T.StructField("program", T.StringType()),
    T.StructField("message", T.StringType()),
])

def load(topic):
    path = f"s3://{raw_bucket}/raw/{topic}/{target.replace('-', '/')}/*"
    src = spark.read.schema(raw_schema).json(path)
    parsed_ts = F.to_timestamp(F.substring("ts", 1, 19), "yyyy-MM-dd'T'HH:mm:ss")
    msg_lower = F.lower(F.col("message"))
    severity = (F.when(msg_lower.rlike(r"\bemerg(ency)?\b"), 0)
        .when(msg_lower.rlike(r"\balert\b"), 1)
        .when(msg_lower.rlike(r"\bcrit(ical)?\b"), 2)
        .when(msg_lower.rlike(r"\b(err|error|failed|failure)\b"), 3)
        .when(msg_lower.rlike(r"\bwarn(ing)?\b"), 4)
        .when(msg_lower.rlike(r"\bnotice\b"), 5)
        .when(msg_lower.rlike(r"\b(info|started|stopped)\b"), 6)
        .when(msg_lower.rlike(r"\bdebug\b"), 7)
        .otherwise(6))
    return (src
        .withColumn("host", F.trim("src_host"))
        .withColumn("ts_fixed", parsed_ts.cast("timestamp_ntz"))
        .withColumn("severity", severity.cast("int"))
        .select("host", F.col("ts_fixed").alias("ts"), "severity",
                "program", F.col("message").alias("msg"),
                F.to_date("ts_fixed").alias("dt"),
                F.hour("ts_fixed").alias("hr"))
        .where("ts IS NOT NULL AND host IS NOT NULL AND host <> ''"))

for topic in ("syslog", "authlog"):
    table = f"glue_prod.logs.{topic}_iceberg"
    df = load(topic).where(F.col("dt") == F.lit(target).cast("date"))
    raw_count = df.count()
    iceberg_count = spark.table(table).where(F.col("dt") == target).count()
    print(f"topic={topic} target={target} raw_count={raw_count} "
          f"iceberg_count={iceberg_count} mode={mode}")

    if mode == "validate":
        if raw_count != iceberg_count:
            raise RuntimeError(f"count mismatch: {topic} {target}")
        continue

    if iceberg_count != 0:
        raise RuntimeError(
            f"refusing to overwrite existing Iceberg data: {topic} {target}")
    df.writeTo(table).append()

severityは既存Viewと同じ判定順（emergencyからdebug、default 6）を移植した。ts - 9 hoursのような補正は行わず、session timezoneをAsia/Tokyoに固定してTIMESTAMP_NTZへ変換する。

4. Iceberg設定で実行する

aws s3 cp raw_to_iceberg.py "s3://$ARTIFACT_BUCKET/jobs/raw_to_iceberg.py" \
  --profile home-aws-admin

EMR_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/raw_to_iceberg.py\",
    \"entryPointArguments\":[\"2026-09-16\",\"validate\",\"${RAW_BUCKET}\"],
    \"sparkSubmitParameters\":\"--conf spark.jars=/usr/share/aws/iceberg/lib/iceberg-spark3-runtime.jar --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\"
  }}" \
  --query jobRunId --output text \
  --profile "${AWS_ADMIN_PROFILE}")

while :; do
  EMR_JOB_STATE=$(aws emr-serverless get-job-run \
    --region "${AWS_REGION}" --application-id "${EMR_APPLICATION_ID}" \
    --job-run-id "${EMR_JOB_RUN_ID}" --query jobRun.state --output text \
    --profile "${AWS_ADMIN_PROFILE}")
  printf 'state=%s\n' "${EMR_JOB_STATE}"
  case "${EMR_JOB_STATE}" in
    SUCCESS) break ;;
    FAILED|CANCELLING|CANCELLED) exit 1 ;;
  esac
  sleep 30
done

最初は必ずvalidateで、既存S3 rawから変換した件数と既存Icebergの件数を比較する。appendを使うのは、オンプレSparkの最終処理日より後で、Iceberg側の対象日が0件であることを確認できた日だけにする。既存日付をDELETEして再作成する処理は、既存Icebergを壊すため本手順では行わない。

5. 確認する

job stateがSUCCESSになり、CloudWatch/S3 logsにparse errorがないことを確認する。Athenaで次を比較する。

SELECT count(*), min(ts), max(ts)
FROM logs.syslog_iceberg
WHERE dt = DATE '2026-09-16';

オンプレIcebergと件数、時刻範囲、host/severity別件数を比較する。不一致ならS3 object key、raw JSON数、invalid timestamp数を順に確認する。

再実行性

既存履歴の検証はread-onlyのvalidate、切り替え後の新規日付だけappendに分けた。同じ日付の再実行は0件確認で停止するため、再処理が必要な場合は対象snapshotと削除範囲を確認した別の運用手順で行う。EventBridge Scheduler等から1日1回起動し、同一日付の同時実行を禁止する。

参考資料

関連記事

実機検証について

本稿は未検証である。特にraw object内のJSON framingとseverity変換は実objectで確認する。

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?