Post Syndicated from Syed Humair original https://aws.amazon.com/blogs/big-data/build-declarative-etl-pipelines-with-aws-glue-6-0/

Data teams commonly build the extract, transform, and load (ETL) pipelines that turn raw order events into analyst-ready aggregates as a bronze, silver, and gold sequence, the medallion architecture. Bronze holds raw ingested records, silver holds cleaned and validated data, and gold holds the business-level aggregates that analysts query. Today you build this on AWS Glue with an orchestrator such as Amazon Managed Workflows for Apache Airflow (Amazon MWAA) or AWS Step Functions coordinating the stages. Many teams run production pipelines exactly this way. As a pipeline grows, the coordination work grows with it: you wire job dependencies, manage intermediate checkpoints, and add retry logic stage by stage.

AWS Glue 6.0, powered by Apache Spark 4.1, introduces Spark Declarative Pipelines (SDP), which simplifies this further. Instead of orchestrating jobs by hand, you declare what each dataset should contain and let the declarative framework resolve dependencies, manage checkpoints, and orchestrate execution order automatically. The result runs as a single declarative job, with no manual directed acyclic graph (DAG) wiring or imperative orchestration code.

In this post, you build a single AWS Glue 6.0 job that turns raw order records into validated, aggregated, analytics-ready tables through the bronze, silver, and gold sequence. You do this without writing any orchestration logic. This walkthrough uses the AWS Command Line Interface (AWS CLI), and the same operations are available through the AWS SDKs.

Solution overview

You build a single AWS Glue 6.0 job that reads raw order records from a CSV file in Amazon Simple Storage Service (Amazon S3). The job flows them through three declared datasets. These are a bronze materialized view (ingest as-is), a silver materialized view (type, validate, and classify), and a gold SQL materialized view (aggregate by region). With AWS Glue Data Catalog integration turned on, all three land as Data Catalog tables, queryable with standard SQL tooling such as Amazon Athena. SDP resolves the dependency order from the dataset references in your code, so you never orchestrate the steps yourself.

Two ways to build the pipeline

Before you build the pipeline, let’s understand this new way of writing ETL pipelines with a quick comparison of the imperative and declarative approaches.

With the imperative approach, you need three AWS Glue jobs, plus an orchestrator to handle sequencing and error handling. A typical pipeline therefore has two layers: an orchestration layer and the ETL processing layer. The following diagram shows this two-layer imperative pipeline.

Figure 1: The two-layer imperative pipeline, with three AWS Glue jobs coordinated by an orchestrator.

Compared to that, the declarative approach runs as a single ETL job with SDP. The following diagram mirrors the previous one, but here it is a single AWS Glue ETL job instead of three jobs plus an orchestrator.

Figure 2: The declarative pipeline, a single AWS Glue job running the bronze, silver, and gold layers with SDP.

The declarative approach reduces more than the number of jobs. It removes the boilerplate that surrounds them. An orchestrator such as Amazon MWAA or AWS Step Functions already handles retries and parallelism, but only at the granularity of a whole job. To get finer control, teams often split a pipeline into several jobs and then hand-wire the dependencies between them. With SDP, you no longer hand-wire a DAG, manage per-stage checkpoints, or split the pipeline into separate jobs for retries and parallelism. SDP derives the dependency graph from your table references and coordinates execution at the level of individual tables. You can still invoke an SDP job from an orchestrator when a broader workflow calls for it, but the pipeline’s internal coordination is no longer code you write and maintain.

SDP separates the what from the how: you declare datasets (the outputs you want), and SDP builds the flows that produce them and runs them as one pipeline, resolving dependencies and execution order automatically.

You declare these abstractions through Python decorators. This post covers three of them, @dp.table, @dp.materialized_view, and @dp.temporary_view, each with its own purpose:

• @dp.table defines a streaming table, which processes new data incrementally on each run. Typical use cases are raw event ingestion and change data capture (CDC) feeds.
• @dp.materialized_view defines a materialized view for batch use cases. Today, this dataset type fully recomputes on each run. Common uses include parsing, aggregations, and machine learning (ML) feature engineering.
• @dp.temporary_view is for temporary computations and aggregations. It’s pipeline-scoped and isn’t persisted outside the pipeline. Use it for enrichment lookups and subqueries.

Streaming tables append only new arrivals. Materialized views fully recompute. This post uses @dp.materialized_view for all three layers to keep the walkthrough focused. In production, you would typically use @dp.table for the bronze layer to process only new files as they arrive rather than re-reading the full source each run.

Running and refreshing the pipeline

When you rerun a pipeline, you don’t always want the same work to happen. Sometimes you only want to confirm the pipeline is well-formed before spending compute. Other times you want to run it but recompute only the datasets that changed rather than the entire graph. SDP handles both cases through two independent controls, and it helps to keep them separate:

• Execution mode (the spark.glue.sdp.jobMode key) answers run or only validate?
• Refresh scope (the spark.glue.sdp.runMode key) answers given that I’m running, what do I recompute?

Execution mode. VALIDATE runs the pipeline in dry-run mode: SDP checks the YAML syntax, dependency resolution, and SQL and Python compilation without writing any data. Use it to verify your pipeline is well-formed before committing compute. RUN (the default) executes the pipeline normally, resolving the dependency graph and materializing datasets.

# Dry run: validate the graph, write nothing
aws glue start-job-run \
  --job-name "${JOB_NAME}" \
  --arguments '{"--conf":"spark.glue.sdp.jobMode=VALIDATE"}' \
  --region "${AWS_REGION}"

# Normal execution
aws glue start-job-run \
  --job-name "${JOB_NAME}" \
  --arguments '{"--conf":"spark.glue.sdp.jobMode=RUN"}' \
  --region "${AWS_REGION}"

Refresh scope. By default, a RUN recomputes every materialized view. You can narrow or widen that with spark.glue.sdp.runMode:

• --refresh <datasets> updates only the named datasets (comma-separated, no spaces).
• --full-refresh <datasets> resets and recomputes only the named datasets (for streaming tables, this also clears their checkpoints).
• --full-refresh-all resets and recomputes every dataset in the pipeline.

# Selective refresh of named datasets
aws glue start-job-run \
  --job-name "${JOB_NAME}" \
  --arguments '{"--conf":"spark.glue.sdp.jobMode=RUN --conf spark.glue.sdp.runMode=--refresh silver_orders,gold_sales_summary"}' \
  --region "${AWS_REGION}"

# Full reset and recompute of the entire pipeline
aws glue start-job-run \
  --job-name "${JOB_NAME}" \
  --arguments '{"--conf":"spark.glue.sdp.jobMode=RUN --conf spark.glue.sdp.runMode=--full-refresh-all"}' \
  --region "${AWS_REGION}"

Selective refresh is useful during development, so you can iterate on a single layer without reprocessing the entire graph. Note that --refresh and --full-refresh each take an explicit list of datasets. To reset the whole pipeline, use --full-refresh-all. Because materialized views hold no incremental state, resetting a materialized view and refreshing it both fully recompute it. The reset-versus-refresh distinction matters for streaming tables, where a refresh processes only new data and a reset clears the checkpoint and reprocesses from scratch.

The multiple values are passed as a single --conf argument string ("spark.glue.sdp.jobMode=RUN --conf spark.glue.sdp.runMode=..."). This is the serialization the AWS Glue SDP mode expects for the run.

Materialized views: Batch transforms with automatic dependency resolution

Materialized views recompute their full result set on each run. SDP infers dependencies from table references: in this pipeline, silver_orders references bronze_orders, so SDP runs bronze first, as shown in the following diagram.

Figure 3: SDP infers the dependency order from table references and runs bronze before silver.

The core pattern is a decorated function that returns a DataFrame:

@dp.materialized_view(comment="Raw orders loaded from CSV")
def bronze_orders() -> DataFrame:
    return spark.read.schema(ORDERS_SCHEMA).option("header", "true").csv(ORDERS_PATH)

The silver layer references bronze_orders through spark.table("bronze_orders"), with no explicit dependency declaration. SDP builds the DAG by analyzing table references in your code and runs bronze first automatically.

Bronze reads every column as a string by design: the bronze layer preserves raw source data without coercion. Type casting, validation, and filtering happen in the silver layer.

SQL and Python coexistence

SDP supports both Python and SQL definitions in the same pipeline project. A SQL materialized view can reference a Python-defined table directly, for example the gold layer aggregating the silver table:

CREATE MATERIALIZED VIEW gold_sales_summary
COMMENT 'Completed-order metrics by region'
AS
SELECT
  region,
  COUNT(*) AS order_count,
  CAST(ROUND(SUM(amount), 2) AS DECIMAL(10, 2)) AS total_sales,
  CAST(ROUND(AVG(amount), 2) AS DECIMAL(10, 2)) AS average_order_value
FROM silver_orders
GROUP BY region;

In this post, Python files define ingestion and validation logic, and SQL files define reporting views and aggregations. SDP discovers both through the libraries glob pattern in the pipeline specification and resolves the cross-language dependencies automatically. The complete source for all three layers follows in the step-by-step walkthrough.

Build the pipeline: Step by step

The rest of this post is a hands-on walkthrough. You build a single AWS Glue 6.0 job that reads orders.csv and processes it through the bronze, silver, and gold layers. The steps are:

1. Prerequisites: AWS account, AWS Identity and Access Management (IAM) role, and S3 bucket.
2. Set up sample data: create orders.csv and upload it to Amazon S3.
3. Build the pipeline files (the spark-pipeline.yml specification plus the three transformation files).
4. Package the pipeline into a zip and upload it to Amazon S3.
5. Create the database: a Data Catalog database with an S3 location.
6. Configure the job: create the AWS Glue 6.0 job with the SDP flag.
7. Validate: run in dry-run mode to verify the graph.
8. Run the pipeline to materialize all datasets.
9. Query results: inspect the tables with Amazon Athena.
10. Clean up: delete the resources you created.

Step 1 – Prerequisites

To follow along, you need:

• An AWS account with access to AWS Glue 6.0.
• A dedicated IAM role trusted by glue.amazonaws.com (set up in the following section).
• A private, encrypted Amazon S3 bucket with Block Public Access enabled.
• The AWS CLI configured with credentials for a non-production account.

IAM role for the pipeline

Create a role that AWS Glue can assume, with the following trust policy:

{
  "Version": "2012-10-17",
  "Statement": [{
    "Effect": "Allow",
    "Principal": { "Service": "glue.amazonaws.com" },
    "Action": "sts:AssumeRole"
  }]
}

Attach the AWS managed policy AWSGlueServiceRole, which grants the AWS Glue Data Catalog and Amazon CloudWatch Logs access the job needs. Then add an inline policy that scopes Amazon S3 access to your bucket, covering the input data, the pipeline zip, the pipeline storage (state) path, and the warehouse location:

{
  "Version": "2012-10-17",
  "Statement": [{
    "Effect": "Allow",
    "Action": ["s3:GetObject", "s3:PutObject", "s3:DeleteObject", "s3:ListBucket"],
    "Resource": [
      "arn:aws:s3:::amzn-s3-demo-bucket",
      "arn:aws:s3:::amzn-s3-demo-bucket/*"
    ]
  }]
}

For a full breakdown of the baseline permissions, see Setting up IAM permissions for AWS Glue.

Set the walkthrough variables

Set the following variables, replacing the example values (us-east-1, amzn-s3-demo-bucket, the account ID 111122223333, and the role name) with your own:

export AWS_REGION="us-east-1"
export BUCKET="amzn-s3-demo-bucket"
export PREFIX="simple-sdp-demo"
export DATABASE="simple_sdp_demo_db"
export ROLE_ARN="arn:aws:iam::111122223333:role/AWSGlueServiceRole-sdp-demo"
export JOB_NAME="simple-sdp-demo"

Step 2 – Set up sample data

The pipeline reads a CSV of order records. Save the following as orders.csv:

order_id,customer_id,region,amount,status,order_ts
O-1001,C-101,EMEA,120.50,COMPLETE,2026-07-23T08:00:00Z
O-1002,C-102,AMER,750.00,COMPLETE,2026-07-23T08:15:00Z
O-1003,C-103,EMEA,-10.00,INVALID,2026-07-23T08:30:00Z
O-1004,C-104,APAC,320.25,COMPLETE,2026-07-23T09:00:00Z
O-1005,C-105,AMER,250.00,COMPLETE,2026-07-23T09:15:00Z
O-1006,C-106,EMEA,90.00,COMPLETE,2026-07-23T09:30:00Z

Upload the file to the input/ location under your project prefix, which is where the bronze layer reads it (the ORDERS_PATH in 01_bronze.py, shown in Step 3). Use the variables you exported in Step 1:

aws s3 cp orders.csv \
  "s3://${BUCKET}/${PREFIX}/input/orders.csv" \
  --region "${AWS_REGION}"

The file includes one invalid order (O-1003, a negative amount), which the silver layer filters out to demonstrate the validation step. The AMER and EMEA regions each have two completed orders, so the gold layer’s order_count and average_order_value are meaningful aggregations rather than single-row passthroughs.

Step 3 – Build the pipeline files

The pipeline project uses the structure introduced earlier: a transformations/ folder holding the three layer definitions (01_bronze.py, 02_silver.py, 03_gold.sql), plus the spark-pipeline.yml specification. The following screenshot shows this layout in a code editor.

Figure 4: The pipeline project layout in a code editor.

The complete contents of each file follow.

3a. spark-pipeline.yml

The specification names the pipeline, points to the Data Catalog database, configures state storage, and discovers transformation files. As with the transformation files, it uses the __DATABASE__, __BUCKET__, and __PREFIX__ tokens, which you substitute at packaging time in Step 4:

name: simple_sdp_demo
catalog: spark_catalog
database: __DATABASE__
storage: s3://__BUCKET__/__PREFIX__/state/
libraries:
  - glob:
      include: transformations/**
configuration:
  spark.sql.shuffle.partitions: "4"

3b. transformations/01_bronze.py

Bronze preserves the raw source as strings. No coercion, no filtering:

"""Bronze layer: preserve source order records as strings."""
from pyspark import pipelines as dp
from pyspark.sql import DataFrame, SparkSession
from pyspark.sql.types import StringType, StructField, StructType

spark = SparkSession.active()

ORDERS_PATH = "s3://__BUCKET__/__PREFIX__/input/orders.csv"

ORDERS_SCHEMA = StructType([
    StructField("order_id", StringType(), True),
    StructField("customer_id", StringType(), True),
    StructField("region", StringType(), True),
    StructField("amount", StringType(), True),
    StructField("status", StringType(), True),
    StructField("order_ts", StringType(), True),
])


@dp.materialized_view(comment="Raw orders loaded from CSV")
def bronze_orders() -> DataFrame:
    return (
        spark.read
        .schema(ORDERS_SCHEMA)
        .option("header", "true")
        .csv(ORDERS_PATH)
    )

The path uses the tokens __BUCKET__ and __PREFIX__ rather than hardcoded values. AWS Glue reads these files from the packaged zip at runtime, so shell variables like ${BUCKET} are not expanded inside them. You substitute the tokens with your real values when you package the project in Step 4, which keeps every file consistent with the variables you exported in Step 1.

3c. transformations/02_silver.py

Silver casts types, filters to complete orders with positive amounts, and derives an amount_band classification:

"""Silver layer: type, validate, and classify complete orders."""
from pyspark import pipelines as dp
from pyspark.sql import DataFrame, SparkSession
from pyspark.sql.functions import col, to_timestamp, trim, when

spark = SparkSession.active()


@dp.materialized_view(comment="Validated complete orders with typed values")
def silver_orders() -> DataFrame:
    typed = (
        spark.table("bronze_orders")
        .select(
            trim(col("order_id")).alias("order_id"),
            trim(col("customer_id")).alias("customer_id"),
            trim(col("region")).alias("region"),
            col("amount").cast("double").alias("amount"),
            trim(col("status")).alias("status"),
            to_timestamp("order_ts", "yyyy-MM-dd'T'HH:mm:ss'Z'").alias("order_ts"),
        )
        .filter(
            col("order_id").isNotNull()
            & col("region").isNotNull()
            & col("order_ts").isNotNull()
            & (col("status") == "COMPLETE")
            & (col("amount") > 0)
        )
    )
    return typed.select(
        "*",
        when(col("amount") >= 500, "large")
        .when(col("amount") >= 100, "medium")
        .otherwise("small")
        .alias("amount_band"),
    )

Silver reads bronze with spark.table("bronze_orders"), so SDP infers the dependency and runs bronze first. Two details matter here:

• The to_timestamp call passes an explicit format, "yyyy-MM-dd'T'HH:mm:ss'Z'". The source timestamps are ISO 8601 with a Z suffix. Giving the format treats Z as a literal and produces the same wall-clock value regardless of the job’s session time zone, which keeps the result deterministic.
• The transformation runs in two projections: the first casts and filters, and the second derives amount_band from the already-typed amount column. Deriving columns with .select(...) rather than a separate .withColumn(...) step keeps SDP’s reference to bronze_orders resolvable as a pipeline dependency. This way, SDP consistently orders the bronze layer before the silver layer. The order matters here too. Spark 4.1 enables ANSI mode by default, so comparing the raw string amount against a number would fail. amount_band therefore reads the already-cast amount.

3d. transformations/03_gold.sql

The gold layer aggregates order metrics by region using SQL:

CREATE MATERIALIZED VIEW gold_sales_summary
COMMENT 'Completed-order metrics by region'
AS
SELECT
  region,
  COUNT(*) AS order_count,
  CAST(ROUND(SUM(amount), 2) AS DECIMAL(10, 2)) AS total_sales,
  CAST(ROUND(AVG(amount), 2) AS DECIMAL(10, 2)) AS average_order_value
FROM silver_orders
GROUP BY region;

Step 4 – Package the project

Substitute the __BUCKET__, __PREFIX__, and __DATABASE__ tokens with the values you exported in Step 1. Then package spark-pipeline.yml and the transformations/ folder into a zip with both at the zip root. Because AWS Glue reads these files from the zip at runtime, the substitution has to happen now, at packaging time, not through shell variables at run time:

# Render the tokens into a build/ copy, leaving your source files untouched
rm -rf build/package && mkdir -p build/package/transformations

sed -e "s|__BUCKET__|${BUCKET}|g" \
    -e "s|__PREFIX__|${PREFIX}|g" \
    -e "s|__DATABASE__|${DATABASE}|g" \
    spark-pipeline.yml > build/package/spark-pipeline.yml

sed -e "s|__BUCKET__|${BUCKET}|g" \
    -e "s|__PREFIX__|${PREFIX}|g" \
    transformations/01_bronze.py > build/package/transformations/01_bronze.py
cp transformations/02_silver.py transformations/03_gold.sql build/package/transformations/

# Zip with the spec and transformations at the zip root
(cd build/package && zip -r -q ../simple-sdp-demo.zip spark-pipeline.yml transformations)

# Upload
aws s3 cp build/simple-sdp-demo.zip "s3://${BUCKET}/${PREFIX}/pipeline/simple-sdp-demo.zip" --region "${AWS_REGION}"

Only spark-pipeline.yml and 01_bronze.py carry tokens, so the other files are copied as-is. The uploaded object is named simple-sdp-demo.zip, which is the same name the job references in Step 6.

Step 5 – Create the database

The database named in spark-pipeline.yml must already exist in the AWS Glue Data Catalog, with an S3 location URI, before the pipeline runs. SDP does not create it automatically:

aws glue get-database --name "${DATABASE}" --region "${AWS_REGION}" >/dev/null 2>&1 \
|| aws glue create-database \
--database-input "{\"Name\":\"${DATABASE}\",\"LocationUri\":\"s3://${BUCKET}/${PREFIX}/warehouse/\"}" \
--region "${AWS_REGION}"

Step 6 – Configure the job

Create an AWS Glue 6.0 job with the zip as ScriptLocation and the SDP flag enabled:

aws glue create-job \
--name "${JOB_NAME}" \
--role "${ROLE_ARN}" \
--command "{\"Name\":\"glueetl\",\"ScriptLocation\":\"s3://${BUCKET}/${PREFIX}/pipeline/simple-sdp-demo.zip\",\"PythonVersion\":\"3\"}" \
--glue-version "6.0" \
--worker-type "G.1X" \
--number-of-workers 2 \
--default-arguments "{\"--enable-spark-declarative-pipeline\":\"true\",\"--enable-glue-datacatalog\":\"true\"}" \
--region "${AWS_REGION}"

Key arguments:

Table 2: Key arguments for the create-job command.

The create-job command sets ScriptLocation to the pipeline zip. You can also point it to an Amazon S3 prefix: upload the unzipped spark-pipeline.yml and transformations/ to a prefix and set ScriptLocation to that prefix (with a trailing /). No other change is needed, and the --enable-spark-declarative-pipeline flag stays the same. The zip keeps the upload to a single object.

Step 7 – Validate (dry run)

Run the job in validation mode first to verify the dependency graph without materializing data:

aws glue start-job-run \
  --job-name "${JOB_NAME}" \
  --arguments '{"--conf":"spark.glue.sdp.jobMode=VALIDATE"}' \
  --region "${AWS_REGION}"

Validation analyzes the project structure, dependency graph, and SQL and Python compilation without creating tables, executing transforms, or writing data. Confirm that the database has no tables after validation completes.

On AWS Glue, validation runs as a job (jobMode=VALIDATE), so you create the job in Step 6 and then validate it here. If you develop locally with the open source spark-pipelines CLI, you can run its dry-run against the project before packaging and uploading.

Step 8 – Run the pipeline

Start the pipeline in normal execution mode:

aws glue start-job-run \
  --job-name "${JOB_NAME}" \
  --arguments '{"--conf":"spark.glue.sdp.jobMode=RUN"}' \
  --region "${AWS_REGION}"

After the run completes, list the materialized tables:

aws glue get-tables \
  --database-name "${DATABASE}" \
  --region "${AWS_REGION}" \
  --query 'TableList[].Name' \
  --output table

Expected tables: bronze_orders, silver_orders, gold_sales_summary.

After the run, the AWS Glue console shows the three output tables in the simple_sdp_demo_db database. The database’s Location is the warehouse path you configured, s3://amzn-s3-demo-bucket/simple-sdp-demo/warehouse/, and each table stores its data under that prefix. The following screenshot shows the database properties and the three tables (bronze_orders, silver_orders, and gold_sales_summary), each registered in the AWS Glue Data Catalog.

Figure 5: The three output tables in the AWS Glue Data Catalog.

Step 9 – Query results

Query the tables with Amazon Athena. If this is your first time using Athena in this Region, set an Amazon S3 query-results location for your workgroup first (Athena console, Settings). Also make sure your identity can read the simple_sdp_demo_db tables in the Data Catalog and the underlying S3 data.

-- Bronze preserves all 6 source rows
SELECT * FROM simple_sdp_demo_db.bronze_orders ORDER BY order_id;

-- Silver retains the 5 complete orders with positive amounts
SELECT * FROM simple_sdp_demo_db.silver_orders ORDER BY order_id;

-- Gold aggregates by region
SELECT * FROM simple_sdp_demo_db.gold_sales_summary ORDER BY region;

Expected gold result:

Table 3: Gold layer aggregation results by region.

Running the query in the Amazon Athena console returns the aggregated result. The following screenshot shows the gold query and its three result rows (AMER, APAC, and EMEA), matching the values in the preceding table.

Figure 6: The gold table results in the Amazon Athena console.

Cost considerations

AWS Glue 6.0 bills ETL jobs by the data processing unit (DPU)-hour, per second, with a 1-minute minimum per run. AWS Glue 6.0 is also priced 30 percent lower per DPU-hour than AWS Glue 5.1, with no change to your workload, so the same job costs less to run on 6.0. This walkthrough runs on 2 G.1X workers (2 DPUs), reads a 6-row CSV, and completes each run in about 2 minutes. It produces three tables in one AWS Glue Data Catalog database.

To estimate the cost of a run, multiply the 2 DPUs by the run time in hours by your Region’s AWS Glue 6.0 DPU-hour rate. You can find that rate on the AWS Glue pricing page, and rates differ by AWS Region. The Amazon S3 objects created are the 6-row CSV, the pipeline zip, and the three tables’ data. To stop further charges, delete the resources when you finish, as shown in the next step.

Step 10 – Clean up

To avoid ongoing charges, delete the resources you created:

# Delete the AWS Glue job
aws glue delete-job --job-name "${JOB_NAME}" --region "${AWS_REGION}"

# Delete the Data Catalog database and its table metadata
aws glue delete-database --name "${DATABASE}" --region "${AWS_REGION}"

# Remove the S3 objects
aws s3 rm "s3://${BUCKET}/${PREFIX}/" --recursive --region "${AWS_REGION}"

What’s next

You now have a single pipeline that turns raw order records into validated, aggregated analytics tables, without writing orchestration logic. From here you can:

• Extend: Add transformation stages (additional @dp.materialized_view functions) and connect them by referencing upstream tables. The pipeline picks up the new dependency automatically.
• Scale: This walkthrough uses materialized views throughout, so every layer fully recomputes on each run (materialized views don’t support incremental refresh). To process only new data as it arrives, convert the bronze layer to a streaming table, which maintains state across runs with checkpoints. For that cross-run state to persist, a streaming table’s data and checkpoint state must not be stored locally. Hive or AWS Glue managed tables require the database’s LocationUri to point to an Amazon S3 path, while Apache Iceberg tables manage their table metadata themselves.
• Govern: Protect the Data Catalog tables SDP produces with AWS Lake Formation fine-grained access control. It enforces table-, row-, column-, and cell-level permissions on read queries in AWS Glue Spark jobs (Glue 5.0 and later, for Hive and Iceberg tables). Because this enforcement covers batch reads, it applies to SDP’s materialized views but not to streaming tables, which read through Spark Structured Streaming.
• Automate: Store the pipeline project in source control. Have your continuous integration and continuous delivery (CI/CD) pipeline package and upload it to Amazon S3 so each job run maps to a known build. Version the zip by object key, or upload the unzipped project to an S3 prefix and turn on Amazon S3 bucket versioning.
• Monitor: Use Amazon CloudWatch metrics and AWS Glue job run insights for pipeline observability, latency tracking, and failure alerting.

Conclusion

In this post, you used Spark Declarative Pipelines, the declarative alternative to explicitly orchestrated ETL, now available in AWS Glue 6.0. Two decorated Python functions and one SQL file define the bronze, silver, and gold datasets, and SDP resolves the dependencies and manages execution order for you.

With SDP, you declare what each dataset should contain and the declarative framework handles ordering and execution. A three-layer pipeline that would otherwise need separate transform and orchestration logic runs as one job that you can ship and maintain.

To get started, open the AWS Glue console and build the walkthrough pipeline, or adapt the pattern to your own bronze, silver, and gold datasets. For the full set of features, see the AWS Glue 6.0 launch announcement. To move existing jobs to the Spark 4.1 runtime, see Upgrade AWS Glue jobs to AWS Glue 6.0 with AI-powered Spark upgrades. For job configuration details, see the AWS Glue Developer Guide.

About the authors

Post Syndicated from Shoukat Ghouse original https://aws.amazon.com/blogs/big-data/build-a-real-time-event-pipeline-with-spark-real-time-mode-on-aws-glue-6-0/

Real-time event pipelines rarely get to work with a uniform schema. Whether it’s IoT metrics, ecommerce clickstreams, or financial pricing vectors, each event type brings its own schema. An equity trade and a rates trade, for instance, carry almost entirely different fields. Ingesting these multi-schema streams has traditionally forced suboptimal architectural choices. You build separate tables for each event type or maintain a wide STRUCT where every possible field across all event types must be declared upfront (fast reads, but sparse and rigid). The other option is to flatten everything into an unwieldy schema with hundreds of columns. To sidestep that maintenance burden, many teams dump events into a plain JSON string column that introduces significant performance penalty. Querying a single nested field requires your engine to deserialize the entire JSON blob for every row. At scale, you burn compute and budget scanning terabytes of raw text to extract a few bytes of data.

Adding to the challenge, these pipelines typically demand mixed processing speeds. You need a real-time path (not near-real-time) to flag anomalies or high-risk events with sub-second latency, while simultaneously pushing those same events into analytical storage for deep historical analysis in batch.

With AWS Glue 6.0, you can tackle all of these challenges (schema heterogeneity, JSON scanning overhead, and mixed-latency requirements) from a single pipeline. Built on Apache Spark 4.1 with Apache Iceberg v3 support, AWS Glue 6.0 brings Variant columns, Variant shredding, Spark Real-Time Mode (RTM), and Arrow-native user-defined functions (UDFs) to a fully managed, serverless environment.

In this post, we walk you through how to build this multi-layer architecture using a financial services use case: a market risk pipeline processing trade pricing vectors. While the example is finance, the patterns apply wherever you deal with heterogeneous schemas, expensive JSON parsing, and mixed real-time/batch requirements such as IoT device fleets, multi-tenant SaaS platforms, logistics tracking, and beyond. We will show you how to flag high-risk trades with sub-second latency, stream everything into an Iceberg v3 data lake as Variants, and run batch Value at Risk (VaR) computations efficiently using Arrow-native UDFs.

Solution overview

A bank’s Market Risk team receives a continuous stream of trade pricing vectors from front-office systems. Each trade event carries:

1. A trade ID and book/desk IDs.
2. A pricing vector as semi-structured data. The schema varies by asset class (for example, equities carry risk sensitivities known as Greeks such as delta/gamma, foreign exchange (FX) carries volatility surfaces, rates carry curve sensitivities).
3. A region ID for jurisdictional reporting (uses a column DEFAULT value, so rows that omit it get auto-populated).

The team needs three things from this stream, each at a different speed. We build the pipeline in three layers, each addressing a distinct requirement with a purpose-built AWS Glue 6.0 capability.

Layer 1: Real-time trade position breach detection (sub-second latency)

Positions must be updated in sub-second time, not seconds of traditional micro-batch streaming. For a team monitoring position limits, those seconds mean trades can breach limits before the system reacts. Spark Real-Time Mode (RTM) eliminates the micro-batch boundary entirely, letting records flow continuously through the pipeline so that high-risk trades trigger alerts within sub-second latency of arrival.

Layer 2: Near-real-time analytical lakehouse (seconds latency)

Every trade must land in a queryable data lake within seconds, with heterogeneous pricing vectors stored without declaring a fixed schema upfront. The Iceberg v3 Variant type handles this natively. The raw semi-structured payload goes into a single column regardless of asset class schema. At write time, Variant shredding automatically extracts fields observed in the data into typed Parquet columns, so downstream analytical queries read only the columns they need without deserializing the full blob. Trade amendments and cancellations are handled at a lower cost with deletion vectors (merge-on-read), and column DEFAULT values reduce boilerplate in ingestion code.

Layer 3: Batch risk computation (minutes to hours latency)

Risk metrics like Value at Risk (VaR) must be computed in Python across millions of trades. Traditional row-by-row pickle serialization between the Java Virtual Machine (JVM) and Python is the bottleneck. Arrow-native UDFs process data as vectorized columnar batches, eliminating serialization overhead and accelerating Python-based risk calculations.

The solution uses three separate AWS Glue 6.0 jobs, each independently scalable:

• Real-time path (Scala, gluestreaming): Reads trades from Amazon Managed Streaming for Apache Kafka (Amazon MSK), enriches them with risk scores and breach flags, and writes alerts to a downstream Kafka topic. It runs with a fixed set of workers that are always on. Downstream fraud detection and position limit systems consume the alerts topic for real-time blocking decisions.
• Near-real-time path (PySpark, gluestreaming): Reads from the same MSK topic and lands the full trade history into an Iceberg v3 table. It uses Glue auto scaling and can scale down between batches, keeping costs lower.
• Batch analytics (PySpark, glueetl): Reads from the Iceberg v3 table, extracts fields using variant_get, computes VaR across the portfolio, and writes aggregated risk reports to a downstream summary table.

The following diagram illustrates the solution architecture.

Figure 1: Real-time market risk pipeline on AWS Glue 6.0

Prerequisites

To follow along with this post, you need the following:

1. An AWS account in a Region where AWS Glue 6.0 is available.
2. An AWS Identity and Access Management (IAM) role with permissions to deploy AWS CloudFormation stacks and create resources including AWS Glue, Amazon MSK, AWS Lambda, Amazon Simple Storage Service (Amazon S3), and the AWS Glue Data Catalog.

Deploy the CloudFormation stack

We provide an AWS CloudFormation template that provisions all the resources needed for this walkthrough.

The stack provisions the following resources:

• An Amazon MSK cluster with two topics: trade-risk-vectors (input) and trade-alerts (real-time alerts output).
• An Amazon S3 bucket for Iceberg table storage and streaming checkpoints.
• An AWS Glue database (risk_analytics_<account-id>_glue6b1).
• An IAM role (GlueRole-<account-id>-glue6b1) with permissions for Glue, MSK, S3, and CloudWatch.
• Virtual private cloud (VPC) networking: A Glue network connection (connection-<account-id>-glue6b1), S3 gateway endpoint, and Glue interface endpoint.
• AWS Glue job rtm-alerts-<account-id>-glue6b1 (Scala): This job reads trades from MSK, scores risk in real time using Spark RTM, writes alerts to the trade-alerts topic.
• AWS Glue job nrt-ingestion-<account-id>-glue6b1 (PySpark): This job reads trades from MSK, writes to Iceberg v3 table with Variant + shredding enabled.
• AWS Glue job batch-var-<account-id>-glue6b1 (PySpark): This job reads from Iceberg v3 table, computes VaR with Arrow UDF, demonstrates deletion vectors.
• AWS Glue job producer-<account-id>-glue6b1-helper (PySpark): This job generates sample trade events (equities, FX, rates) to the trade-risk-vectors topic.

Deploy the CloudFormation stack:

1. Download the CloudFormation template from the GitHub repository.
2. Sign in to the AWS CloudFormation console
3. Choose Create stack > With new resources > Upload a template file, and upload the downloaded template.
4. Enter the following parameters:
   • VpcId: Your VPC ID.
   • SubnetIds: At least two subnets in different Availability Zones.
   • SecurityGroupId: A dedicated security group that allows all inbound TCP traffic from itself (self-referencing rule).
   • RouteTableId: The main route table for your VPC.
5. Acknowledge the IAM capabilities and choose Create stack.

Stack creation takes approximately 20 minutes.

After the stack completes, open the AWS Glue console and start the jobs in this order:

1. Start rtm-alerts-<account-id>-glue6b1 and nrt-ingestion-<account-id>-glue6b1.
2. Once both show RUNNING, start producer-<account-id>-glue6b1-helper.
3. After the producer finishes (~3.5 minutes), run batch-var-<account-id>-glue6b1 for risk aggregation.

The consumers must be running before the producer starts so that trades are scored in real time and landed in the Iceberg table as they arrive. The batch job runs last because it reads from the Iceberg table that the near-real-time path populates.

Understand the Iceberg v3 table design

The CloudFormation stack provisions Glue jobs that create two Iceberg v3 tables, trade_risk_vectors (primary trade store) and daily_risk_summary (batch VaR output), using new data types and features:

1. VARIANT: Stores semi-structured pricing vectors without requiring a fixed schema.
2. DEFAULT values: Automatically applies provided defaults when fields aren’t provided.
3. Deletion vectors (merge-on-read): Enables fast row-level updates and deletes.

Open the AWS Glue console under Data Catalog > Tables > trade_risk_vectors.

Figure 2: The trade_risk_vectors table in the AWS Glue Data Catalog

The following is the Create Table command:

CREATE TABLE {TABLE} (
    trade_id STRING, book_id STRING, desk STRING,
    asset_class STRING DEFAULT 'UNKNOWN',
    execution_time STRING,
    pricing_vector VARIANT,
    var_contribution DOUBLE DEFAULT 0.0,
    risk_weight DOUBLE DEFAULT 1.0,
    trade_date DATE, region STRING DEFAULT 'EMEA'
) USING iceberg
TBLPROPERTIES ('format-version'='3', 'write.delete.mode'='merge-on-read',
    'write.update.mode'='merge-on-read',
    'write.parquet.shred-variants'='true')
PARTITIONED BY (trade_date, asset_class)

Note the use of DEFAULT values for asset_class, var_contribution, risk_weight, and region. This is an Iceberg v3 feature that applies defaults automatically when values aren’t provided during writes, reducing boilerplate in ingestion code. The pricing_vector column is defined as a Variant type, and write.parquet.shred-variants='true' automatically extracts Variant fields into separate typed Parquet columns at write time for faster downstream queries.

Sample trade event generator

The CloudFormation stack includes a Glue job (producer-<accountid>-glue6b1-helper) that produces realistic trade events to the trade-risk-vectors MSK topic. Each event carries a pricing_vector with a completely different schema per asset class. This is exactly the problem Variant solves.

Equity trade (greeks, scenarios with sector/region breakdowns):

Figure 3: Sample equity trade pricing vector

Rates trade (curve sensitivities per tenor, calibration params):

Figure 4: Sample rates trade pricing vector

Completely different structures: greeks vs curve sensitivities, BlackScholes vs HullWhite. Both land in the same pricing_vector VARIANT column with no schema changes required.

Ingest trades with Spark Real-Time Mode

Traditional Spark Structured Streaming uses micro-batches: collect records, schedule a job, process, commit, wait. Even with small batches, the fixed overhead of planning and scheduling adds noticeable latency per batch. For a risk team monitoring position limits, the delay can let a trade breach a limit before the system reacts.

The following Scala job reads trade events from Amazon MSK, applies lightweight risk rules based on data directly available in the event, and writes alerts to a Kafka topic, all with sub-second latency. The real-time path intentionally avoids external lookups (market data, volatility surfaces) to stay fast. The full VaR computation happens later in the batch layer where latency is less critical.

You can view the complete job code in the AWS Glue console under the rtm-alerts-<accountid>-glue6b1 job. Additionally, all the scripts are available in the GitHub repository.

Figure 5: Scala real-time job that scores trades and writes alerts

The Trigger.RealTime("1 minute") is what distinguishes this from a traditional micro-batch. Records flow through the pipeline continuously. Records are processed the instant they arrive. The 1-minute parameter controls how often Spark checkpoints its progress for recovery. It does not control how often records are processed. RTM on AWS Glue 6.0 currently supports Kafka-source, stateless, Scala workloads with fixed workers (no auto scaling) and update output mode only. This makes it ideal for stateless transformations that require sub-second latency, such as the filter, enrich, score, and route pattern shown here. The heavier computation (VaR, aggregations) runs in the micro-batch/batch layer where sub-second latency is less critical.

The real-time path acts as a circuit breaker: trades over $50M notional are flagged CRITICAL, over $25M flagged HIGH. Downstream systems consume the trade-alerts topic and can block or escalate before the next trade executes. The detailed VaR computation (which requires market data, volatility surfaces, and the full pricing vector) runs in the batch consumption layer where latency is less sensitive.

After the streaming phase completes, the job reads back from the trade-alerts topic and measures end-to-end latency. It compares two MSK timestamps: when the trade was received by MSK from the producer, and when the alert was received by MSK from RTM.

To verify the alerts and latency, open the Amazon CloudWatch console > Log groups > /aws-glue/jobs/output and select the RTM job’s log stream. You will see the alert summary showing each flagged trade with its end-to-end latency. The following is a sample.

Figure 6: CloudWatch output showing flagged trades and end-to-end latency

Store trades in Iceberg v3 with Variant shredding enabled

The near-real-time path reads from the same MSK topic but writes to an Iceberg v3 table using standard micro-batch streaming. This job runs separately with auto scaling enabled, scaling between batches, keeping costs lower than the always-on real-time path.

You can view the complete job code in the AWS Glue console under the nrt-ingestion-<accountId>-glue6b1 job. The critical aspects are the Variant conversion and the Iceberg write:

Figure 7: Near-real-time PySpark job writing trades to Iceberg v3 as a Variant

The PARSE_JSON() function converts the raw pricing vector into a native Variant, regardless of the asset class schema. Whether the incoming trade is an equity with greeks, an FX option with a volatility surface, or a rates swap with curve sensitivities, it all goes into the same column. Since the table has write.parquet.shred-variants enabled, fields observed in the initial sample are automatically extracted into typed Parquet columns for fast downstream queries.

How shredding works

During the write process, Spark automatically extracts the Variant fields it observes into separate typed Parquet columns at write time, a feature called shredding. At the start of each write, the engine buffers a sample of rows (controlled by write.parquet.variant-inference-buffer-size), infers which fields exist and their types, then uses that schema to shred all subsequent rows in the file. Every field observed in that sample gets its own typed column, including nested objects. Rows that lack a particular field simply store NULL in that shredded column. For our risk table, fields like $.greeks.delta, $.dv01, and $.model all live in their own typed Parquet columns, even if only one asset class carries a specific field. The result: faster read performance because queries access only the typed columns they need, skipping the rest of the document entirely. Shredding is transparent to queries. variant_get() calls work the same way whether the field is shredded or not. The query engine automatically routes to the shredded column when available, falling back to the binary Variant blob for fields that aren’t part of the inferred schema.

Note: Shredding adds write latency because the engine must infer the schema and write additional typed columns. In this pipeline, we enable shredding on the near-real-time path and absorb that cost, since the downstream read benefits (batch VaR, ad-hoc queries, audit) far outweigh the write penalty. For latency-sensitive pipelines where every millisecond on the write path matters, you can disable shredding on the streaming table and instead write shredded data in a separate batch job that reads from the unshredded table and inserts into a shredded copy. This approach trades architectural simplicity for lower ingestion latency.

Build the batch consumption layer

The third AWS Glue 6.0 job reads from the Iceberg v3 table, extracts risk metrics from the Variant column, computes VaR using an Arrow-native UDF, and writes aggregated results to a summary table.

You can view the complete job code in the AWS Glue console under the batch-var-<accountid>-glue6b1 job. The key aspects are the variant_get extraction from deeply nested structures and the Arrow-native UDF:

Figure 8: Extracting nested Variant fields with variant_get

Notice how variant_get reaches into arbitrarily nested structures: $.greeks.delta (2 levels), $.scenarios[0].breakdown.by_sector.financials (5 levels), $.model_params.calibration.fit_error (4 levels). All with the same function call. No pre-flattening, no schema-per-asset-class tables, no ETL to restructure the data before querying.

Once the risk metrics are extracted, we need to run a Monte Carlo-style Historical VaR that simulates 1,000 daily profit and loss (P&L) scenarios per trade and returns the 99th percentile loss. This is where the @arrow_udf decorator comes in.

Figure 9: Arrow-native UDF computing Historical VaR

The @arrow_udf decorator is new in Spark 4.1. Your function receives and returns pyarrow.Array directly, operating on the entire batch of rows at once. There is no pickle serialization, no row-by-row invocation, and no Pandas conversion. Data flows as native Arrow columnar arrays between the JVM and Python. For compute-heavy operations like VaR across hundreds of thousands of rows, this can be significantly faster than traditional scalar UDFs. Additionally, you can use the built-in UDF profilers to identify performance and memory bottlenecks in compute-heavy UDFs such as VaR calculations.

Handle late trade corrections with deletion vectors

In financial markets, trade amendments and cancellations are common. The batch VaR job demonstrates this after completing the risk computation. It amends one trade and cancels another.

Figure 10: Amending and cancelling trades with merge-on-read

Prior to Iceberg v3, row-level deletes required either rewriting entire data files (copy-on-write) or maintaining separate positional delete files that store (file_path, row_position) pairs as Parquet rows (merge-on-read). Both approaches are expensive at scale. Copy-on-write rewrites gigabytes for a single amendment, and positional deletes degrade read performance as delete files accumulate (each read must parse and hash-join all delete records against the data file).

Because we configured the table with write.delete.mode='merge-on-read', UPDATEs and DELETEs write deletion vectors instead of positional delete files used in Iceberg v2. A deletion vector is a Roaring Bitmap stored in a Puffin file (.puffin), one per affected data file, marking which row positions are deleted. At read time, the engine loads a single bitmap and skips flagged positions with a bit check. No file joins, no linear scan through multiple delete files. The bitmap is compact regardless of how many rows are deleted, and read performance remains predictable as amendments accumulate.

The batch job also verifies the deletion vectors were created. You can see the results in the job’s output logs.

Figure 11: Output verifying deletion vectors were created

Clean up

To avoid incurring further charges, delete the CloudFormation stack. This removes all resources provisioned as part of this post, including the S3 bucket, Glue jobs, Iceberg tables, MSK cluster, and IAM roles.

Conclusion

In this post, we built a multi-layer market risk pipeline using AWS Glue 6.0 (real-time alerting, near-real-time ingestion, and batch analytics):

• Spark Real-Time Mode (RTM) on the real-time path delivers sub-second trade scoring and breach alerting, eliminating the micro-batch boundary so position limits are enforced before the next trade executes.
• Iceberg v3 Variant on the near-real-time path stores heterogeneous pricing vectors without schema flattening. One table handles equities, FX, and rates with different schemas per row.
• Variant shredding delivers faster reads by automatically extracting fields into separate typed Parquet columns at write time with no manual tuning required.
• Arrow-native UDFs eliminate pickle serialization overhead for Python-based risk calculations, processing data as vectorized columnar batches on the batch layer.
• Deletion vectors handle trade amendments and cancellations without costly data file rewrites, using compact Roaring Bitmaps instead of accumulating positional delete files.
• Default values reduce boilerplate in ingestion code.

To get started with AWS Glue 6.0, see the AWS Glue documentation. For more information about Apache Iceberg v3, see the Iceberg specification.

About the authors