S3メタデータの活用方法を考えてみた(S3×Glue×Athenaでの構成)

Last updated at 2025-01-23Posted at 2025-01-23

はじめに

こんにちは。株式会社ジールの@yakisobapanです。

AWS re:Invent 2024で発表されたS3メタデータの活用方法を考えてみました。

アップデート内容

Amazon S3にオブジェクトが格納、変更、削除された際に生成されるメタデータを取得できるようになりました。
メタデータは、同タイミングで発表されたS3テーブルにて保持され、Athena等を利用してクエリすることが可能です。

Amazon S3 メタデータ (プレビュー) の発表 — メタデータを管理する最も簡単かつ迅速な方法
（2024/12/03 What's New with AWS?より）

Amazon S3 テーブルの発表 — 分析ワークロード向けに最適化されたフルマネージド型の Apache Iceberg テーブル
（2024/12/03 What's New with AWS?より）

東京リージョンでも利用可能になっていました！

Amazon S3 テーブルが新たに 5 つの AWS リージョンで利用可能になりました
（2025/1/17 What's New with AWS?より）

アーキテクチャ

以下のような構成で考えてみました。
一般的なAWS上でのETL構成*に今回のアップデートを活用できそうです。

*S3をデータストアとして、データソース(csv等)をGlueジョブでParquetへ変換し格納、Athenaでクエリするといった構成

赤矢印が新機能のS3メタデータに関するデータフロー、参照フローです。

やってみた

事前準備
上記構成を作成します。

S3
・「テーブルバケット」形式でバケットを作成します。
　⇒末尾「参考」のドキュメントをご参照ください
・バケット配下にinフォルダ、outフォルダを作成します。
・データソースとなるcsvファイルをinフォルダに配置します。
Glue
・各用途のData Catalogを作成します。
・ジョブを作成し、csvをParquetへ変換するPythonスクリプトを記述します。

transform_csv_to_parquet

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

# setup
args = getResolvedOptions(sys.argv, ['JOB_NAME'])
sc = SparkContext()
glueContext = GlueContext(sc)
spark = glueContext.spark_session
job = Job(glueContext)
job.init(args['JOB_NAME'], args)

# data read
input_path = "s3://yakisobapan-bucket-s3-metadata/in/bread/"
output_path = "s3://yakisobapan-bucket-s3-metadata/out/bread/"

datasource0 = glueContext.create_dynamic_frame.from_options(
    format_options={"withHeader": True, "separator": ","},
    connection_type="s3",
    format="csv",
    connection_options={"paths": [input_path]},
    transformation_ctx="datasource0"
)

# transform Parquet(Snappy)
datasink1 = glueContext.write_dynamic_frame.from_options(
    frame=datasource0,
    connection_type="s3",
    connection_options={"path": output_path},
    format="parquet",
    format_options={"compression": "SNAPPY"}
)

job.commit()