Tuần 4 - Ngày 1: Glue ETL sâu — Spark và DynamicFrame

Mục tiêu học tập

• Ôn mô hình thực thi Spark: driver, executor, partition, lazy evaluation
• Hiểu DynamicFrame vs DataFrame và các transform đặc trưng của Glue
• Xử lý schema không đồng nhất: ResolveChoice, Relationalize
• Ghi output đúng cách: partitionKeys, update catalog, các format

1. Spark trong 10 dòng (nền tảng để hiểu Glue)

• Lazy evaluation: transformation (filter, map, join) chỉ được ghi vào DAG; action (write, count) mới kích hoạt chạy
• Shuffle: join/groupBy đòi trộn dữ liệu giữa executor — tốn nhất; skew (khóa lệch) gây 1 task chạy mãi
• Số partition ~ mức song song; file quá ít/quá to → ít task; quá nhiều file nhỏ → overhead

2. DynamicFrame vs DataFrame

DynamicFrame là abstraction riêng của Glue trên Spark, thiết kế cho dữ liệu schema không sạch/không đồng nhất:

# Chuyển qua lại khi cần API của cả hai
df = dynamic_frame.toDF()              # DynamicFrame → DataFrame (Spark SQL)
dyf = DynamicFrame.fromDF(df, glueContext, "dyf")

Thực dụng: đọc/ghi + resolve schema bằng DynamicFrame, transform phức tạp bằng DataFrame/Spark SQL.

3. Các transform DynamicFrame hay thi

# Ví dụ: cột "price" lúc là int, lúc là string trong JSON nguồn
resolved = raw_dyf.resolveChoice(specs=[("price", "cast:double")])

# JSON nested → các bảng phẳng cho Redshift
flat = Relationalize.apply(frame=raw_dyf, staging_path=tmp_dir, name="root")

Exam cue: "JSON lồng nhau nhiều tầng cần nạp vào Redshift" → Relationalize. "Cột có kiểu dữ liệu không nhất quán" → ResolveChoice.

4. Đọc và ghi dữ liệu trong Glue job

Đọc

# Từ catalog (khuyến nghị — dùng được bookmark, partition pushdown)
dyf = glueContext.create_dynamic_frame.from_catalog(
    database="sales_db", table_name="orders",
    push_down_predicate="year='2026' AND month='07'",   # chỉ đọc partition cần!
    transformation_ctx="src")

• push_down_predicate: lọc partition trước khi đọc S3 — giảm mạnh thời gian/chi phí (khác filter sau khi đọc)

Ghi

sink = glueContext.getSink(
    connection_type="s3", path="s3://lake/curated/orders/",
    updateBehavior="UPDATE_IN_DATABASE",     # cập nhật schema vào catalog
    partitionKeys=["year", "month"],
    enableUpdateCatalog=True,
    transformation_ctx="sink")
sink.setFormat("glueparquet")               # parquet tối ưu ghi của Glue
sink.writeFrame(transformed_dyf)

• enableUpdateCatalog + partitionKeys: partition mới tự vào catalog — không cần crawler chạy sau job
• Ghi Iceberg/Hudi/Delta: dùng Spark DataFrame writer với cấu hình table format (Glue 4.0/5.0 có sẵn thư viện)

5. Job nhỏ mà nguồn JDBC — lưu ý parallel read

Đọc bảng JDBC lớn: mặc định 1 kết nối → chậm. Cấu hình hashfield/hashpartitions (DynamicFrame) hoặc partitionColumn/lowerBound/upperBound/numPartitions (Spark) để đọc song song.

Câu hỏi ôn tập

1. Nguồn JSON có cột user_id khi là số khi là chuỗi. Glue xử lý thế nào?

   Xem đáp án

   DynamicFrame giữ cột đó thành choice type (không fail). Dùng ResolveChoice để chốt: cast:string (ép về một kiểu), make_cols (tách thành user_id_int, user_id_string), make_struct, hoặc project. Đây là lợi thế chính của DynamicFrame so với DataFrame vốn đòi schema thống nhất.

2. Cần nạp JSON nested (order chứa mảng items) vào các bảng quan hệ trong Redshift. Transform nào của Glue?

   Xem đáp án

   Relationalize — làm phẳng cấu trúc nested thành bảng root + các bảng con cho từng mảng, nối bằng khóa sinh tự động. Kết quả là tập bảng phẳng nạp thẳng vào Redshift (COPY) và join lại được.

3. Job chỉ cần dữ liệu tháng hiện tại từ bảng partition theo year/month nhưng đang đọc cả bucket rồi mới filter. Sửa thế nào?

   Xem đáp án

   Dùng push_down_predicate khi create_dynamic_frame.from_catalog (vd "year='2026' AND month='07'") — Glue chỉ liệt kê và đọc các partition khớp trên S3, thay vì đọc hết rồi filter trong Spark. Với bảng rất nhiều partition có thể dùng thêm catalogPartitionPredicate (server-side). Giảm cả runtime lẫn DPU-hour.

4. Sau job ghi partition mới vào S3, Athena không thấy dữ liệu mới. Cách sửa ngay trong job (không thêm crawler)?

   Xem đáp án

   Ghi bằng sink có enableUpdateCatalog=True kèm partitionKeys (và updateBehavior) — Glue tự thêm partition/schema mới vào Data Catalog ngay khi ghi. Phương án khác: gọi MSCK REPAIR TABLE/ALTER TABLE ADD PARTITION qua Athena, hoặc dùng partition projection.

5. Khi nào nên chuyển DynamicFrame sang DataFrame trong job?

   Xem đáp án

   Khi cần API Spark đầy đủ: Spark SQL, window functions, join phức tạp, UDF, thư viện ML — DataFrame phong phú và thường nhanh hơn (Catalyst optimizer). Pattern chuẩn: DynamicFrame để đọc + resolve schema + bookmark, toDF() để transform nặng, fromDF() quay lại để ghi bằng Glue sink (update catalog).

6. Đọc bảng PostgreSQL 500GB qua JDBC rất chậm dù job nhiều worker. Nguyên nhân?

   Xem đáp án

   Mặc định JDBC đọc một kết nối duy nhất — mọi worker khác ngồi chờ. Cấu hình đọc song song: hashfield/hashpartitions với DynamicFrame (hoặc partitionColumn, lowerBound, upperBound, numPartitions với Spark) trên một cột phân bố đều (khóa số/timestamp). Cân nhắc cả tải lên database nguồn.

Bài tập thực hành

• Viết Glue job đọc JSON nested mẫu, chạy Relationalize và xem các bảng output
• Tạo file có cột kiểu lẫn lộn, quan sát choice type và thử 3 chiến lược ResolveChoice
• Bật enableUpdateCatalog trong sink, ghi partition mới, verify Athena thấy ngay không cần crawler
• Đọc Glue PySpark transforms reference

Tài liệu tham khảo chính thức

• DynamicFrame class
• Relationalize
• Creating tables, updating schema from jobs
• Reading from JDBC in parallel

---

Tiếp theo: Glue Job Tuning và tối ưu hiệu năng