Where do bad records go?

Do not silently drop them. Write rejected rows to a separate S3 location with a reason column, and monitor how many there are. A sudden rise in rejects usually means the source changed.

AWS Glue Module 5: Transforms and Cleaning Data

Introduction

Most Glue work is cleaning and shaping data. Glue provides built-in transforms for common tasks, and you can mix them with Spark SQL and functions. This module covers the transforms you will use most, with the sample orders data and a customers file.

Prerequisites

Module 4 completed

Step-by-Step Guide

Step 1: Rename and cast columns with ApplyMapping

ApplyMapping takes a list of tuples: source column, source type, target column, target type. It selects only the columns you list, which also removes the ones you do not need.

mapped = ApplyMapping.apply(
    frame=orders,
    mappings=[
        ("order_id", "long", "order_id", "int"),
        ("customer_id", "long", "customer_id", "int"),
        ("order_date", "string", "order_date", "date"),
        ("amount", "double", "amount", "double"),
        ("status", "string", "status", "string"),
    ],
    transformation_ctx="mapped",
)

Step 2: Resolve choice types

When a column contains mixed types, Glue records a choice type. ResolveChoice decides what to do: cast to one type, make a struct, project, or split into separate columns. Resolving choices early prevents errors when writing to typed formats such as Parquet.

resolved = ResolveChoice.apply(
    frame=mapped,
    specs=[("amount", "cast:double")],
    transformation_ctx="resolved",
)

Step 3: Drop nulls and filter rows

DropNullFields removes fields that are null in every row. Filter keeps rows that satisfy a function. For clearer logic on larger jobs, many people convert to a DataFrame and use Spark's filter and column expressions.

cleaned = DropNullFields.apply(frame=resolved, transformation_ctx="cleaned")

shipped = Filter.apply(
    frame=cleaned,
    f=lambda row: row["status"] != "CANCELLED" and row["amount"] > 0,
    transformation_ctx="shipped",
)

Step 4: Join datasets

Join combines two DynamicFrames on key columns. For large joins, the Spark DataFrame join gives more control, such as join type and broadcast hints for small lookup tables.

customers = glueContext.create_dynamic_frame.from_catalog(
    database="glue_lab", table_name="raw_customers", transformation_ctx="customers_source"
)

joined = Join.apply(shipped, customers, "customer_id", "id")

# DataFrame version with a broadcast hint for the small table
from pyspark.sql.functions import broadcast
df_joined = shipped.toDF().join(broadcast(customers.toDF()), shipped.toDF().customer_id == customers.toDF().id, "left")

Step 5: Standardize and deduplicate with Spark functions

Use Spark SQL functions for text cleanup, null handling, and removing duplicates. Decide how to pick the row to keep when duplicates differ, for example the latest by a timestamp.

from pyspark.sql import functions as F
from pyspark.sql.window import Window

df = joined.toDF()
df = (df
      .withColumn("city", F.initcap(F.trim(F.col("city"))))
      .withColumn("amount", F.coalesce(F.col("amount"), F.lit(0.0))))

latest = Window.partitionBy("order_id").orderBy(F.col("updated_at").desc())
deduped = (df.withColumn("rn", F.row_number().over(latest))
             .filter("rn = 1").drop("rn"))

FAQs

Where do bad records go?: Do not silently drop them. Write rejected rows to a separate S3 location with a reason column, and monitor how many there are. A sudden rise in rejects usually means the source changed.

Summary

ApplyMapping renames and casts, ResolveChoice fixes mixed types, DropNullFields and Filter remove bad data, Join combines tables, and Spark functions handle text cleanup and deduplication. Keep rejected records for review.