How Custom Transformations Work in AWS Glue and PySpark

Every Glue job eventually needs logic that no built-in transform provides: a bespoke parsing rule, a domain-specific validation, a scoring model. The question is never whether you can write it — you always can — but where you put it. That choice routinely swings job runtime by an order of magnitude, and it is almost always made by accident.

This is a companion to Building Production Data Pipelines with AWS Glue and PySpark. That piece covered the platform; this one goes into the part people actually get wrong.

Four places custom logic can live

A Glue job has four distinct extension points, and they are not interchangeable:

Where	What it operates on	Runs in
DynamicFrame transforms	One record at a time, as a Python dict	Python worker
Glue Studio custom node	A whole `DynamicFrameCollection`	Driver, then executors
Spark UDFs	A column value, or a batch of them	Python worker
Native Spark expressions	Whole columns, compiled	JVM executor

The last one is not really a "custom transformation" at all, which is exactly why it gets overlooked. It is also the fastest by a wide margin, and a surprising amount of supposedly-custom logic can be expressed with it.

The built-in DynamicFrame transforms

Glue ships a set of transforms designed for messy, schema-inconsistent input — the situation Spark's DataFrame handles badly. Worth knowing before you write anything yourself:

Transform	What it does
`ApplyMapping`	Renames and casts fields; drops anything unlisted
`ResolveChoice`	Settles a field that arrives as more than one type
`Relationalize`	Flattens nested structures into a set of joinable flat frames
`Unbox`	Parses a string field containing JSON/XML into real structure
`Unnest`	Lifts nested struct fields to the top level
`SelectFields` / `DropFields`	Project columns in or out
`SplitRows` / `SplitFields`	Split one frame into a collection of two
`Join`	Join two DynamicFrames without converting to DataFrame

Note

Relationalize is the one people miss. Given deeply nested JSON it returns a DynamicFrameCollection — a root frame plus one per repeated array, with generated keys to join them back. It turns "we cannot load this JSON into a warehouse" into a normalization problem you already know how to solve.

Writing your own row-level transform

Map and Filter take an arbitrary Python function. Each record arrives as a plain dict, which makes them forgiving of inconsistent schemas — a missing key is just a missing key.

DynamicFrame Map and Filter

from awsglue.transforms import Map, Filter

COUNTRY_ALIASES = {"usa": "US", "u.s.": "US", "united states": "US"}

def normalize_record(rec):
    # rec is a plain dict. Missing keys are normal in raw data, so use .get().
    email = (rec.get("email") or "").strip().lower()
    rec["email"] = email or None

    country = (rec.get("country") or "").strip().lower()
    rec["country"] = COUNTRY_ALIASES.get(country, country.upper() or None)

    rec["is_corporate"] = bool(email) and not email.endswith(
        (".gmail.com", "@gmail.com", "@yahoo.com", "@hotmail.com")
    )
    return rec

cleaned = Map.apply(
    frame=raw_dyf,
    f=normalize_record,
    transformation_ctx="normalize_records",
)

valid = Filter.apply(
    frame=cleaned,
    f=lambda r: r.get("order_total") is not None and r["order_total"] >= 0,
    transformation_ctx="drop_invalid_orders",
)

This is the most forgiving option and the slowest. Every record crosses from the JVM into a Python process and back. On a few million rows that is fine. On a few billion it dominates the job.

Why row-at-a-time Python is expensive

Spark runs on the JVM. Your Python code does not. Anything written in Python has to cross a process boundary, and how often it crosses is the whole story.

The same work. The difference is how many times the data is serialized.

A row-at-a-time UDF pickles each row, writes it to a socket, wakes a Python interpreter, gets a result back, and unpickles it. A vectorized UDF hands over an Arrow record batch — typically around 10,000 rows — in a columnar format both sides already understand, with no per-row serialization at all.

There is a second cost that is easy to miss: Catalyst cannot see inside any Python function. A native expression can be reordered, folded into a projection, or pushed down into the scan. A UDF is an opaque box the optimizer must call as written, so filters below it cannot move above it.

First choice: native expressions

Before writing any UDF, check whether pyspark.sql.functions already does it. The library is much larger than most people use, and the coverage of string, date, JSON, array and map operations is extensive.

Things that look custom but are not

from pyspark.sql import functions as F

df = (df
    # string cleanup — no UDF needed
    .withColumn("email", F.lower(F.trim("email")))
    .withColumn("domain", F.regexp_extract("email", r"@(.+)$", 1))

    # conditional logic — no UDF needed
    .withColumn("tier", F.when(F.col("revenue") >= 100000, "enterprise")
                         .when(F.col("revenue") >= 10000,  "mid_market")
                         .otherwise("smb"))

    # nested JSON in a string column — no UDF needed
    .withColumn("payload", F.from_json("payload_raw", "struct<id:string,tags:array<string>>"))
    .withColumn("tag_count", F.size("payload.tags"))

    # windowed logic — no UDF needed
    .withColumn("prev_order_ts",
                F.lag("order_ts").over(Window.partitionBy("user_id").orderBy("order_ts")))
)

Note

In practice most "we need a UDF" requests we review turn out to be expressible natively. It is worth ten minutes of searching the function list before writing one — the native version is faster, optimizable, and needs no schema declaration.

Python UDFs

When logic genuinely cannot be expressed natively, a plain UDF is the simplest escape hatch. Declare the return type explicitly; without it Spark assumes StringType.

A Python UDF that earns its keep

from pyspark.sql.functions import udf
from pyspark.sql.types import StructType, StructField, StringType, BooleanType

import phonenumbers            # shipped via --additional-python-modules

PHONE = StructType([
    StructField("e164",    StringType()),
    StructField("region",  StringType()),
    StructField("is_valid", BooleanType()),
])

@udf(returnType=PHONE)
def parse_phone(raw, default_region):
    if not raw:
        return None
    try:
        n = phonenumbers.parse(raw, default_region or "US")
    except phonenumbers.NumberParseException:
        return ("", "", False)
    return (
        phonenumbers.format_number(n, phonenumbers.PhoneNumberFormat.E164),
        phonenumbers.region_code_for_number(n) or "",
        phonenumbers.is_valid_number(n),
    )

df = df.withColumn("phone", parse_phone("phone_raw", "country"))

Never let a UDF raise

A UDF must return None for null input and never raise. One unhandled exception fails the task, which fails the stage, which fails the job — after retries, and usually hours in. Wrap the body defensively and emit a sentinel you can count later.

pandas UDFs, and the batch trap

A pandas UDF receives an Arrow batch as a pd.Series and returns one of the same length. Same interface from Spark's perspective, dramatically less overhead.

Series to Series

from pyspark.sql.functions import pandas_udf
import pandas as pd

@pandas_udf("string")
def normalize_sku(s: pd.Series) -> pd.Series:
    # Vectorized pandas operations — one call per ~10,000 rows, not per row.
    return (s.str.strip()
             .str.upper()
             .str.replace(r"[^A-Z0-9]", "", regex=True)
             .str.slice(0, 12))

df = df.withColumn("sku", normalize_sku("sku_raw"))

Where a batch needs expensive one-off setup — loading a model, opening a lookup file — use the iterator form. The setup runs once per partition rather than once per batch:

Iterator of Series — amortise expensive setup

from typing import Iterator

@pandas_udf("double")
def score(batches: Iterator[pd.Series]) -> Iterator[pd.Series]:
    model = load_model_from_s3()      # once per partition
    for batch in batches:
        yield pd.Series(model.predict_proba(batch.values.reshape(-1, 1))[:, 1])

df = df.withColumn("propensity", score("feature_value"))

The batch trap

A pandas UDF sees one batch, not the whole column. So (v - v.mean()) / v.std() inside a pandas UDF computes the mean of a few thousand arbitrary rows, not of your dataset — and returns confidently wrong numbers with no error. Any calculation that depends on the full column belongs in a window function or an aggregation, not a pandas UDF.

Grouped work: applyInPandas and mapInPandas

When your logic needs a whole group at once — sessionisation, interpolation, per-entity model fitting — applyInPandas hands you each group as a complete pandas DataFrame.

applyInPandas — a whole group at a time

SESSION_SCHEMA = "user_id string, event_ts timestamp, page string, session_id long"

def sessionise(pdf: pd.DataFrame) -> pd.DataFrame:
    pdf = pdf.sort_values("event_ts")
    gap_s = pdf["event_ts"].diff().dt.total_seconds().fillna(0)
    pdf["session_id"] = (gap_s > 1800).cumsum()      # 30-minute inactivity window
    return pdf

sessions = (df
    .groupBy("user_id")
    .applyInPandas(sessionise, schema=SESSION_SCHEMA))

Groups must fit in memory

Each group must fit in one executor's memory. A skewed key — a bot account with 40 million events — will OOM that task while every other task idles. Check your group-size distribution before reaching for this, and treat a wide distribution as a signal to use a window function instead.

mapInPandas is the ungrouped sibling: it streams batches of the whole DataFrame and lets you return a different number of rows than you received — useful for explode-style logic that native functions cannot express.

Glue Studio custom transform nodes

If a job is built visually, the Custom Transform node drops a Python function into the middle of the graph. The signature is fixed and catches people out: it receives and returns a DynamicFrameCollection, not a frame.

Glue Studio custom transform node

def MyTransform(glueContext, dfc) -> DynamicFrameCollection:
    # dfc is a collection even when there is exactly one input.
    name = list(dfc.keys())[0]
    df = dfc.select(name).toDF()

    enriched = (df
        .withColumn("order_date", F.to_date("order_ts"))
        .withColumn("margin", F.col("revenue") - F.col("cost")))

    # Returning several frames gives the node several output ports.
    high = enriched.filter(F.col("margin") >= 0)
    bad  = enriched.filter(F.col("margin") < 0)

    return DynamicFrameCollection({
        "clean":     DynamicFrame.fromDF(high, glueContext, "clean"),
        "quarantine": DynamicFrame.fromDF(bad,  glueContext, "quarantine"),
    }, glueContext)

Returning multiple frames is the genuinely useful part: it is how you split a stream into a clean path and a quarantine path inside the visual editor, without dropping to a fully code-based job.

Registering a reusable visual transform

Custom nodes written inline are invisible to every other job. Glue lets you register a transform once and have it appear in the Studio palette for the whole account — two files in the Glue assets bucket:

The transform itself

# s3://aws-glue-assets-<account>-<region>/transforms/hash_pii.py

def hash_pii(self, columns, salt_param):
    from pyspark.sql import functions as F
    df = self.toDF()
    for col in [c.strip() for c in columns.split(",")]:
        df = df.withColumn(col, F.sha2(F.concat(F.lit(salt_param), F.col(col).cast("string")), 256))
    return DynamicFrame.fromDF(df, self.glue_ctx, "hashed")

DynamicFrame.hash_pii = hash_pii

hash_pii.json — the palette entry beside it

{
  "name": "hash_pii",
  "displayName": "Hash PII columns",
  "description": "SHA-256 hashes the named columns with a per-environment salt.",
  "functionName": "hash_pii",
  "path": "s3://aws-glue-assets-123456789012-us-east-1/transforms/hash_pii.py",
  "parameters": [
    { "name": "columns",    "displayName": "Columns (comma separated)", "type": "str", "isOptional": false },
    { "name": "salt_param", "displayName": "Salt",                      "type": "str", "isOptional": false }
  ]
}

Drop both files in the bucket and the transform shows up in Glue Studio for anyone in the account. For rules that must be applied identically everywhere — PII hashing, currency normalization, tenant tagging — this is much safer than trusting each team to copy the snippet correctly. Check the current AWS documentation for the exact config schema before you build against it; the field set has grown over time.

Packaging a shared transform library

Past a handful of jobs, transformations belong in a versioned package rather than pasted into each script.

Job parameters

# Third-party packages, resolved by pip at job start
--additional-python-modules   phonenumbers==8.13.40,pydantic==2.7.1

# Your own library, built as a wheel and published to S3
--extra-py-files              s3://delantech-artifacts/libs/delantech_transforms-0.4.2-py3-none-any.whl

# Useful on constrained workers
--python-modules-installer-option  --no-cache-dir

Pin every version. An unpinned --additional-python-modules resolves at job start, which means a job that ran correctly last night can fail tonight because a dependency published a release — a genuinely miserable failure to debug at 3am.

Custom transforms and job bookmarks

Bookmarks track progress per transformation_ctx. Two rules follow, and breaking either causes silent data problems rather than errors:

Give every DynamicFrame transform a stable transformation_ctx. Changing the string resets that step's bookmark and reprocesses everything.
Converting to a DataFrame drops out of bookmark tracking entirely. Once you call .toDF(), Glue no longer tracks what was consumed. Keep the bookmarked read at the boundary — on the create_dynamic_frame call — and do custom work after it.

Bookmark at the edges, custom logic in the middle

raw = glueContext.create_dynamic_frame.from_catalog(
    database="analytics", table_name="raw_orders",
    transformation_ctx="raw_orders_source",     # bookmarked here
)

df = raw.toDF()                                  # from here on, not tracked
df = apply_custom_transformations(df)            # your logic

glueContext.write_dynamic_frame.from_options(
    frame=DynamicFrame.fromDF(df, glueContext, "out"),
    connection_type="s3",
    connection_options={"path": target, "partitionKeys": ["order_date"]},
    format="glueparquet",
    transformation_ctx="orders_sink",            # and here
)
job.commit()

Testing transformations

The single highest-value structural decision: keep transformations as pure functions over DataFrames, importing nothing from awsglue. They then run on a plain local Spark session, in seconds, with no AWS involved.

Pure functions are testable anywhere

# delantech_transforms/orders.py — no Glue imports
from pyspark.sql import functions as F

def add_margin(df):
    return df.withColumn("margin", F.col("revenue") - F.col("cost"))

# tests/test_orders.py
def test_add_margin_handles_null_cost(spark):
    df = spark.createDataFrame(
        [(100.0, 40.0), (100.0, None)], "revenue double, cost double")
    out = {r["revenue"]: r["margin"] for r in add_margin(df).collect()}
    assert out[100.0] in (60.0, None)     # null cost propagates, does not crash

Row-level dict functions from Map.apply are even easier — they are ordinary Python and need no Spark at all:

No Spark required

def test_normalize_record_lowercases_email():
    assert normalize_record({"email": "  A@B.COM "})["email"] == "a@b.com"

def test_normalize_record_survives_missing_keys():
    assert normalize_record({})["email"] is None

Choosing: a decision tree

In order. Stop at the first "yes".

Can pyspark.sql.functions express it? → Use that. Fastest, optimizable, no schema to declare.
Does it need a whole group at once? → applyInPandas, after checking group sizes for skew.
Is it row-independent and vectorizable? → pandas UDF. Iterator form if setup is expensive.
Does it change the row count? → mapInPandas, or explode if the shape allows.
Is the schema too messy for a DataFrame? → DynamicFrame.Map, at ingest only, then convert.
None of the above? → Python UDF, defensively written.

Rough magnitudes we measure on real workloads. Treat them as ordering, not as numbers to quote — the spread depends heavily on how much work the function does per row:

Approach	Typical relative cost	Catalyst sees inside?
Native `F.*` expressions	1× (baseline)	Yes
pandas UDF (Series → Series)	~2–5×	No
`applyInPandas`	~5–15×	No
Python UDF (row at a time)	~10–100×	No
`DynamicFrame.Map`	comparable to a Python UDF	No

Note

On one clickstream job we reviewed, a single Python UDF doing string normalization accounted for 71% of runtime. Rewriting it as three chained native expressions — no pandas, no UDF — took twenty minutes and cut the job from 48 minutes to 14. The lesson is not that UDFs are bad; it is that the fastest custom transformation is usually the one you discover you did not need.

If you are still deciding how these jobs get scheduled and chained together, that is a separate question with its own trade-offs: Apache Airflow vs AWS Glue.

Running into these problems on your own platform?

We do free data platform assessments — we map your pipelines, surface quick wins, and hand you a 90-day engineering roadmap.

Book an assessment