Day17 淬鍊之章-Glue 實作篇-2

2025 iThome 鐵人賽

DAY 17

Build on AWS

動漫宅的 30 天 AWS Lakehouse 修行日誌系列第 17 篇

17th鐵人賽 aws

小田

團隊三陳牛肉吉事堡

2025-10-01 23:55:07

73 瀏覽

分享至

簡介

在上篇 Day16 淬鍊之章-Glue 實作篇-1 中，我們實作了 Glue 所需的 IAM Role 建立和指派，以及實際建立一支 Glue PySpark Job。

本篇我們要來建立另一支 Glue PySpark Job 來處理 Silver animes 與 ratings 的 Table Join，並使用 Glue Workflow 將 silver Job 與 gold job 串接成一個 Pipeline。

Iceberg 在 S3 的儲存結構

首先，我們先來確認當我們在 Glue + Iceberg 中建立表格時，S3 會產生的兩個主要目錄，例如：

s3://anime-lake/gold.db/animes_summary/
├── data/ # 實際資料檔案 (Parquet, ORC, Avro)
└── metadata/ # Iceberg 表的元資料 (JSON, Avro)

📂 `data/` 目錄

用途：存放實際的資料檔案（通常是 Parquet）。

檔案結構：

每個 partition 對應到一個子目錄，例如：

data/year=2021/genres=Action/part-0000-xxxx.parquet
data/year=2021/genres=Comedy/part-0001-yyyy.parquet
data/year=0/genres=Unknown/part-0002-zzzz.parquet

檔名裡帶有唯一 ID，保證不可變（immutable file）。

重點：Iceberg 採用 append-only 模型，不會覆蓋檔案，而是寫新檔再更新 metadata。

📂 `metadata/` 目錄

用途：追蹤表的狀態、欄位 schema、partition 設定、歷史版本。
檔案類型：
- v0001.metadata.json
  定義 schema、partition 規則，以及目前有哪些 snapshot。
- snapshots.avro
  每次寫入會建立一個 snapshot，指向當前活躍的 data files。
- manifest.avro / manifest-list.avro
  記錄有哪些 parquet 檔屬於這個 snapshot。
重點：
- metadata 檔非常小，但極度重要。
- Athena、Spark 查詢 Iceberg 表時，會先讀取 metadata/，再決定要掃哪些 data/ 檔案。
- 這就是 Iceberg 能夠支援 Schema Evolution（欄位新增/刪除）和 Time Travel（查歷史版本）的基礎。

小結

data/ = 真正的資料檔案（依 partition 存放）。
metadata/ = Iceberg 的腦袋，記錄「哪些檔案屬於這張表、schema 是什麼、目前最新版本是哪個 snapshot」。

這也是為什麼有時候 Glue Job 寫入失敗會看到 metadata 殘留 → commit abort，因為 Iceberg 在設計上強調 一致性 (Consistency)，metadata 一旦不完整，就會中止寫入。

建立 Gold Job

了解完 Iceberg 的分區儲存設計後，我們來實際建立 Glue Job

Step1：

首先我們一樣來建立一支 Glue Job 並命名為 gold_anime_summary
並將 code 填入下方的 script 區塊內
本次我們在 gold 的 job 設計上將 silver.animes 和 silver.rating 做 Join，並採用 year + genres 作為 GroupBy 條件和 Partition 設定
接著取得 anime_count、avg_anime_score、total_ratings、avg_user_rating 等指標

from awsglue.context import GlueContext
from pyspark.context import SparkContext
from pyspark.sql import functions as F
from awsglue.job import Job

# 初始化 Glue Job
sc = SparkContext()
glueContext = GlueContext(sc)
spark = glueContext.spark_session
job = Job(glueContext)

# Iceberg Catalog 設定
spark.conf.set("spark.sql.catalog.glue_catalog", "org.apache.iceberg.spark.SparkCatalog")
spark.conf.set("spark.sql.catalog.glue_catalog.warehouse", "s3://anime-lake/")
spark.conf.set("spark.sql.catalog.glue_catalog.catalog-impl", "org.apache.iceberg.aws.glue.GlueCatalog")
spark.conf.set("spark.sql.catalog.glue_catalog.io-impl", "org.apache.iceberg.aws.s3.S3FileIO")

# 建立 Gold Database
spark.sql("CREATE DATABASE IF NOT EXISTS gold LOCATION 's3://anime-lake/gold.db/'")

# 讀取 Silver 層資料
silver_animes = spark.table("glue_catalog.silver.animes")
silver_ratings = spark.table("glue_catalog.silver.ratings")

# ====== year 欄位清理：確保 partition 安全 ======
# year：非 4 位數 → 0
silver_animes = silver_animes.withColumn(
    "year",
    F.when(F.col("year").rlike("^[0-9]{4}$"), F.col("year").cast("int"))
     .otherwise(F.lit(0))
)

# genres：NULL 或空字串 → 'Unknown'
silver_animes = silver_animes.withColumn(
    "genres",
    F.when(F.col("genres").isNull() | (F.trim(F.col("genres")) == ""), F.lit("Unknown"))
     .otherwise(F.trim(F.col("genres")))
)

# ====== Join + 聚合 ======
gold_summary = (
    silver_animes.join(
        silver_ratings,
        silver_animes["animeID"] == silver_ratings["anime_id"],
        "left"
    )
    .groupBy("year", "genres")
    .agg(
        F.countDistinct("animeID").alias("anime_count"),
        F.avg("score").alias("avg_anime_score"),
        F.count("rating").alias("total_ratings"),
        F.avg("rating").alias("avg_user_rating")
    )
)

# ====== 寫入 Gold (Iceberg Table) ======
gold_summary.write \
    .format("iceberg") \
    .partitionBy("year", "genres") \
    .mode("overwrite") \
    .saveAsTable("glue_catalog.gold.animes_summary")

job.commit()

Step2：設定 Job detials