AWS Glueでデータ変換パイプラインを構築する方法｜クローラー・ETLジョブ・Parquet変換・スケジュール実行

AWS Glue でデータ変換パイプラインを作る基本形は、「S3 に置いた生データをクローラーでデータカタログに登録する → ETL ジョブで整形して Parquet などの形式で別の S3 に書き出す → トリガーで定期実行する」という流れです。この記事では、CSV や JSON のログを分析しやすい形に変換するパイプラインを例に、IAM ロールの準備からジョブのスクリプト、増分処理、実行の監視までを説明します。

Glue はバージョン（Glue 4.0、5.0 など）ごとに使える Spark や Python のバージョンが異なり、Glue Studio の画面も更新が続いています（2026年10月現在）。画面の項目名は目安として読み、使うバージョンの仕様は AWS の公式ドキュメントで確かめてください。

パイプラインの全体像

Glue の主な部品は次のとおりです。

• データカタログ：データの場所とスキーマ（列名と型）を記録する台帳です。データベースとテーブルの単位で管理し、Athena や Redshift Spectrum からも参照されます。
• クローラー：S3 などのデータを読み取ってスキーマを推定し、データカタログにテーブルを作成・更新します。
• ETL ジョブ：データを読み込んで変換し、書き出す処理です。Spark ベースのジョブが中心で、小さな処理向けの Python シェルジョブもあります。Glue Studio のビジュアルエディタで組み立てることも、スクリプトを直接書くこともできます。
• トリガー / ワークフロー：ジョブやクローラーを、スケジュールや前の処理の完了をきっかけに起動します。

ここでは次の構成を作ります。

1. s3://example-raw/access_logs/dt=2026-10-01/ のように日付ごとに置かれた CSV ログ
2. クローラーで raw_db.access_logs テーブルを作成
3. ETL ジョブで型を整え、日付でパーティションを分けた Parquet として s3://example-curated/access_logs/ に書き出す
4. 書き出したデータを Athena で集計する

dt=2026-10-01 のような「列名=値」の形のフォルダー名にしておくと、クローラーが dt をパーティション列として認識します。後で日付を指定したクエリが、その日のフォルダーだけを読むようになるので、処理量を抑えられます。

ステップ1：IAM ロールを作る

Glue のクローラーとジョブは、IAM ロールの権限で S3 を読み書きします。信頼されたエンティティを Glue にしたロールを作り、次の権限を付けます。

• AWS 管理ポリシー AWSGlueServiceRole（Glue の基本動作に必要な権限）
• 読み込み元バケットへの s3:GetObject と s3:ListBucket
• 書き込み先バケットへの s3:PutObject と、必要に応じて s3:DeleteObject

S3 の権限は対象のバケットとプレフィックスに絞ります。s3:* をすべてのリソースに付けると、ジョブのスクリプトに誤りがあったとき、関係の無いデータまで書き換えられる危険があります。バケットを KMS で暗号化している場合は、その鍵の利用権限も必要です。

ステップ2：クローラーでテーブルを作る

コンソールでクローラーを作成し、データソースに s3://example-raw/access_logs/、IAM ロールに先ほどのロール、出力先にデータベース raw_db を指定します。作成後に実行すると、データカタログにテーブルができます。

aws glue start-crawler --name raw-access-logs-crawler
aws glue get-crawler --name raw-access-logs-crawler --query "Crawler.State"
aws glue get-table --database-name raw_db --name access_logs

クローラーの結果は推定なので、作成されたテーブルの列と型を確認します。CSV に見出し行が無い、列ごとに値の形式がばらばらといった場合、列名が col0 のようになったり、すべて文字列型になったりします。推定が合わないときは、分類子（classifier）を設定するか、型の変換を ETL ジョブ側で行います。

ステップ3：ETL ジョブを書く

Glue Studio でスクリプトのジョブを作り、IAM ロールと Glue のバージョンを選んでから、次のようなスクリプトを書きます。データカタログのテーブルを読み、列の型を変換し、dt でパーティションを分けた Parquet として書き出します。

import sys
from awsglue.transforms import ApplyMapping
from awsglue.utils import getResolvedOptions
from awsglue.context import GlueContext
from awsglue.job import Job
from pyspark.context import SparkContext

args = getResolvedOptions(sys.argv, ["JOB_NAME", "output_path"])

sc = SparkContext()
glueContext = GlueContext(sc)
job = Job(glueContext)
job.init(args["JOB_NAME"], args)

# データカタログのテーブルを読み込む
src = glueContext.create_dynamic_frame.from_catalog(
    database="raw_db",
    table_name="access_logs",
    transformation_ctx="src",
)

# 列名と型を整える（必要な列だけ残す）
mapped = ApplyMapping.apply(
    frame=src,
    mappings=[
        ("request_time", "string", "request_time", "string"),
        ("path", "string", "path", "string"),
        ("status", "string", "status", "int"),
        ("bytes", "string", "bytes", "long"),
        ("dt", "string", "dt", "string"),
    ],
    transformation_ctx="mapped",
)

# dt ごとにフォルダーを分けて Parquet で書き出す
glueContext.write_dynamic_frame.from_options(
    frame=mapped,
    connection_type="s3",
    connection_options={"path": args["output_path"], "partitionKeys": ["dt"]},
    format="parquet",
    transformation_ctx="sink",
)

job.commit()

output_path はジョブのパラメーターで渡します。ジョブの設定で --output_path に s3://example-curated/access_logs/ を指定するか、実行時に引数で渡します。書き出し先をスクリプトに直接書かず、パラメーターにしておくと、検証用と本番用で同じスクリプトを使い回せます。

Parquet は列単位で圧縮して保存する形式で、Athena で一部の列だけを集計するときに読み込む量が CSV より大幅に減ります。分析用のデータは Parquet にしておくのが一般的です。

ステップ4：増分処理を有効にする（ジョブブックマーク）

毎日実行するジョブで、毎回すべてのデータを読み直すと、日を追うごとに処理時間が伸びます。ジョブブックマークを有効にすると、前回の実行で処理済みのファイルを記録し、新しく追加された分だけを読むようになります。

ジョブブックマークはジョブの設定で有効にするか、パラメーター --job-bookmark-option に job-bookmark-enable を指定します。正しく動かすには、スクリプト側で次の2点が欠かせません。

• 読み込み処理に transformation_ctx を付ける
• 最後に job.commit() を呼ぶ

job.commit() の前にエラーで止まった場合、その回の処理は記録されず、次回にもう一度読まれます。書き出し先に同じデータが重複しないよう、失敗したときは書き出し先の該当パーティションを確認してから再実行します。

ステップ5：実行と監視

aws glue start-job-run \
  --job-name logs-to-parquet \
  --arguments '{"--output_path":"s3://example-curated/access_logs/"}'

aws glue get-job-runs --job-name logs-to-parquet --max-items 5 \
  --query "JobRuns[].{id:Id,state:JobRunState,error:ErrorMessage}"

ジョブのログは CloudWatch Logs に出力されます。標準出力は /aws-glue/jobs/output、エラーは /aws-glue/jobs/error のロググループが基本です。失敗したときは、ジョブ実行の画面に出るエラーメッセージを確認し、詳細はこれらのログで Python の例外や Spark のエラーを探します。よくある原因は次の順です。

1. 権限不足：S3 の AccessDenied。IAM ロールに対象バケットの権限があるか、KMS の権限があるかを確認します。
2. スキーマの不一致：元データに想定外の列や値が入り、型の変換に失敗する。クローラーの結果と実データを見比べます。
3. メモリ不足：大きなファイルを扱うとワーカーのメモリが足りなくなることがあります。ワーカーの種類を大きくするか、入力ファイルを分割します。

ジョブにはタイムアウトと最大同時実行数を明示しておくと、想定外に長く動き続けて費用がかさむ事故を防げます。急がない処理では、起動に時間がかかる代わりにコストを抑える Flex 実行クラスも選べます。

定期実行にする

決まった時刻に動かすだけなら、スケジュール型のトリガーで cron 式を指定します。「クローラーが終わったらジョブを動かす」のように処理をつなげたい場合は、Glue のワークフローで条件付きトリガーを組むか、Step Functions から順番に呼び出します。

変換後のデータを使う

書き出し先にもクローラーを実行するか、テーブル定義を作成すれば、Athena から SQL で集計できます。WordPress サイトのアクセスログや、WooCommerce から書き出した注文データを日次で整形しておくと、本番のデータベースに負荷をかけずに分析できます。本番の MySQL に直接つないで抽出する場合は、読み取り専用のユーザーやリードレプリカを使い、アクセスの少ない時間帯に動かします。

まとめ

• Glue のパイプラインは、クローラーでデータカタログに登録し、ETL ジョブで変換し、トリガーで定期実行する構成が基本です。
• IAM ロールは AWSGlueServiceRole に加え、S3 の権限を対象のバケットとプレフィックスに絞って付けます。
• 分析用のデータは Parquet にし、dt= 形式のパーティションで読む量を抑えます。
• 増分処理はジョブブックマークで行い、transformation_ctx と job.commit() を忘れないようにします。
• 失敗時は CloudWatch Logs を見て、権限、スキーマ、メモリの順に原因を確認します。

• AWS S3ストレージの設定方法
• AWS IAMユーザーとロールの使い分け