Amazon EMR 原生集成 AWS Glue 数据质量

Bad data rarely announces itself. It slips into an Amazon EMR pipeline quietly. A null where there should be a value. A timestamp in the wrong format. A referential key that no longer exists. It travels downstream until it surfaces as a wrong number in a dashboard, a failed downstream job, or a business decision made on data nobody verified.

For teams running large-scale data processing on Amazon EMR, choosing a data quality strategy has meant picking between two trade-offs. You can build your own quality checks in Apache Spark using an open source library like Deequ. With this approach, you can customize checks to your exact pipeline, but it adds custom code your team must maintain, with no centralized ruleset management and no consistent quality catalog across pipelines. Or you can route your data through AWS Glue to access its managed data quality capabilities. This approach minimizes the code you need to maintain but turns data quality validation into a separate step and adds another engine to your pipeline.

Today, there is a third option.

Amazon EMR Data Quality built on AWS Glue is now available natively on Amazon EMR, bundled in AWS runtime for Apache Spark (emr-spark-8.1.0) and later versions. You can write data quality rules using the same Data Quality Definition Language (DQDL) used in AWS Glue. Run them directly inside your Amazon EMR Spark jobs, across each of the three Amazon EMR deployment modes: Amazon EMR on EC2, Amazon EMR Serverless, and Amazon EMR on EKS.

This post walks through two patterns for using AWS Glue Data Quality, a capability of AWS Glue, on Amazon EMR. The first shows a lightweight ad hoc validation workflow, useful for exploratory processing and pipeline-local checks. The second builds on the Apache Iceberg Write-Audit-Publish (WAP) pattern. An extract, transform, and load (ETL) job writes data to an Iceberg branch, retrieves a managed ruleset from the AWS Glue Data Catalog, and validates the branch. Only if the data passes your quality bar does it merge to the main table. By the end of this post, you will have working examples of both patterns ready to run on any mode of Amazon EMR.

What is Amazon EMR Data Quality?

With Amazon EMR Data Quality, you can define and run data quality rules against your datasets. At its core is the Data Quality Definition Language (DQDL), a declarative rule syntax. With DQDL, you can express constraints like completeness, uniqueness, value ranges, referential integrity, and statistical distributions in plain, readable statements.

DQDL also includes machine learning (ML)-based anomaly detection. Rather than setting explicit thresholds, DetectAnomalies rules learn the statistical baseline of your data across runs and flag deviations automatically. This is useful for metrics that shift over time, like daily record counts or revenue totals, where a fixed range rule would require constant maintenance.

You can define rules inline within a job or store them as managed rulesets in the AWS Glue Data Catalog, a centralized metadata repository for table definitions and data quality rulesets. After the rules are stored, you can version these rulesets, share them across teams, and reuse them across multiple pipelines. The evaluator returns per-rule pass or fail results and an overall quality score. Your job can inspect the score to take action: fail the pipeline, write results to Amazon Simple Storage Service (Amazon S3), trigger an alert, or conditionally merge data.

You can now run this rule engine directly inside your Spark jobs on Amazon EMR. The same DQDL rulesets you author for AWS Glue work without modification.

For a full reference on DQDL syntax and rule types, see the AWS Glue Data Quality documentation.

Prerequisites and setup

Amazon EMR is a managed service for running Apache Spark and other open source big data frameworks on AWS. It offers three deployment modes: Amazon EMR on EC2, Amazon EMR Serverless, and Amazon EMR on EKS.

Before running the examples in this post, make sure you have the following in place.

Amazon EMR release: Amazon EMR Data Quality is bundled in AWS runtime for Apache Spark (emr-spark-8.1.0) and later versions across Amazon EMR on EC2, Amazon EMR on EKS, and Amazon EMR Serverless. No bootstrap action or additional installation is required. The library is available as part of the release.

AWS Identity and Access Management (IAM) permissions: The execution role for your Amazon EMR job needs the following permissions:

• glue:AccessDataQualityRuntimeConfiguration to publish data quality results to the AWS Glue Data Catalog.
• glue:CreateDataQualityRuleset to create managed rulesets in the AWS Glue Data Catalog.
• glue:GetDataQualityRuleset to retrieve managed rulesets from the AWS Glue Data Catalog.
• glue:GetDataQualityResult to read prior quality results.
• glue:PublishDataQuality to submit data quality results to the AWS Glue Data Catalog.
• glue:GetDatabase and glue:GetTable for catalog access.
• Standard Amazon S3 read/write permissions for your data and output locations.

Pattern 1: Ad hoc data quality validation

This pattern defines data quality rules directly inside your Amazon EMR Spark job, with no externally managed ruleset to maintain.

The use case

Your team ingests a raw CSV file from a third-party vendor and needs to validate it before loading it into your data lake. The rules are straightforward: check that the record ID is never null, that revenue values are non-negative, and that country codes conform to an allowed list. This is a one-off check that lives entirely within the job. There is no centralized ruleset to manage, and the goal is to validate data quality before the load.

This pattern works well for:

• Exploratory or one time processing pipelines.
• Early-stage pipelines where rules are still evolving.
• Lightweight validation that does not require centralized governance.

How it works

You define your DQDL rules inline as a string, pass your DataFrame and the rules to EvaluateDataQuality, and inspect the results to decide how the job should proceed.

To validate data quality inline using PySpark

To implement this pattern using PySpark, see the following code sample, which can be run on all three Amazon EMR deployment modes:

import os
from pyspark.sql import SparkSession
from awsdataquality.transforms import EvaluateDataQuality

spark = SparkSession.builder \
    .appName("dq-ruleset-catalog") \
    .enableHiveSupport() \
    .getOrCreate()

# Read new incoming data
input_df = spark.read.parquet("s3://amzn-s3-demo-sales-data/sales_data/incoming_data/")

# Contains our DQDL rules and analyzers. Analyzers submit statistics to the AWS Glue service for inspection. This can help create new rules.
ruleset = """
Rules = [
    RowCount > 0,
    IsPrimaryKey "order_id",
    Completeness "customer_email" >= 0.95,
    ColumnValues "order_status" in ["PENDING","SHIPPED","DELIVERED","RETURNED"],
    ColumnValues "net_amount" between 0.0 and 25000.0
]
Analyzers = [
    AllStatistics "net_amount",
    DistinctValuesCount "country",
    RowCount
]
"""

# Publishing options for the results of the data quality evaluation
publishing_opts = {
    "dataQualityEvaluationContext": "blog-post-dq-evaluation-context",
    "enableDataQualityResultsPublishing": "true"
}

# AWS Glue Data Quality uses the additional_options parameter to configure advanced evaluation behaviors, output formats, anomaly detection and performance tuning for data quality checks.
additional_opts = {"compositeRuleEvaluation.method": "ROW"}

# This set tells the Data Quality Evaluator if there are other data sources that
# we need to pull in for checks such as cross table reference checking.
additional_sources = {}

# Evaluate the ruleset against the data frame created by reading the parquet files and save
# the results in dq_results as a DataFrame.
dq_results = EvaluateDataQuality.apply(input_df, ruleset, publishing_opts, additional_opts,
    additional_sources)

print(f"{dq_results.count()} rule result row(s)")  # forces evaluation

# Iterate through the DQ results for any failed rules and if there are any failed
# rules, throw an exception to stop the job.
failures = []
for row in dq_results.collect():
    if row["Outcome"] == "Failed":
        failures.append(f"Rule: {row['Rule']} | Reason: {row['FailureReason']}")

if failures:
    failure_summary = "\n".join(failures)
    raise RuntimeError(f"Data quality checks failed:\n{failure_summary}")

# Write out data into the Iceberg table sales_data table
input_df.writeTo("company_sales.sales_data").append()
print(f"all rules passed; appending {input_df.count()} row(s) to company_sales.sales_data")

The preceding script passes the incoming DataFrame, an inline ruleset, and publishing options to EvaluateDataQuality, which runs the ruleset and returns a results DataFrame. The results DataFrame lists each rule with a Passed or Failed outcome and, for failed rules, the reason. The publishing options control if and where the results get published. The script raises an error and stops if any rule fails. If every rule passes, it appends the incoming data to the company_sales.sales_data table.

What the data quality results look like

The EvaluateDataQuality.apply() function returns the results as a DataFrame containing each rule, whether it passed or failed, and the evaluated rule. Your job can read this DataFrame and branch its logic accordingly: writing clean records to Amazon S3, routing failed records to a quarantine path, or halting the pipeline entirely. Following is sample output of the code earlier:

RuntimeError: Data quality checks failed:
Rule: ColumnValues "order_status" in ["PENDING","SHIPPED","DELIVERED","RETURNED"] | Reason: Value: 0.99598 ...
Rule: ColumnValues "net_amount" between 0.0 and 25000.0 | Reason: Value: -2228.87 ...

Pattern 2: Iceberg Write-Audit-Publish with managed rulesets

This pattern gates data promotion on a centrally managed ruleset, merging new data into your main table only after it passes your quality checks.

The use case

Your team runs a nightly ETL job that transforms raw transactional data and writes it to an Apache Iceberg table. Apache Iceberg is an open table format for large analytic datasets that supports schema evolution, time travel, and branching on Amazon S3. Rather than writing directly to the production table and discovering quality issues after the fact, you want a safety gate. Write to a staging branch, validate it against your organization’s data quality standards, and only promote the data to the main table if it passes.

This is the Write-Audit-Publish (WAP) pattern, a well-established practice in Iceberg-based architectures. For a detailed walkthrough of this pattern using AWS Glue, see Build Write-Audit-Publish pattern with Apache Iceberg branching and AWS Glue Data Quality. WAP gives you atomic promotion: either all the new data meets your quality bar and gets merged, or none of it does. Readers of the main table should not see partial or failed writes.

When combining AWS Glue Data Quality with WAP, you define your quality rules once in the AWS Glue Data Catalog and have your data governance team own them. You enforce them automatically at promotion time, without touching the pipeline code. Combined with managed rulesets in the AWS Glue Data Catalog, you get quality enforcement that is consistent and auditable. The data quality rules are also decoupled from the job code and changing them does not require touching the pipeline.

This pattern works well for:

• Production ETL pipelines where bad data reaching the main table has downstream consequences.
• Organizations with centralized data governance teams that own and version quality standards.
• Apache Iceberg-based lakehouses where the WAP pattern is already in use.

How it works

The flow has four steps:

• Retrieve the ruleset – Call the AWS Glue Data Catalog to fetch the managed ruleset for this table. The ruleset is owned by your governance team and versioned in the catalog.
• Write to a branch – Before writing the ETL output, create an Apache Iceberg branch on the target table. Write all transformed records to this branch rather than to main.
• Evaluate – Run EvaluateDataQuality against the branch data using the retrieved ruleset.
• Merge or discard – If the quality score meets your threshold, merge the branch into main. If not, discard the branch and trigger your alerting workflow.

Creating a managed ruleset: To create a ruleset in the AWS Glue Data Catalog that targets the same database and table, you can create one using the AWS Command Line Interface (AWS CLI). For example:

aws glue create-data-quality-ruleset \
    --name "sales-orders-ruleset" \
    --ruleset "Rules = [ RowCount > 0, IsPrimaryKey \"order_id\", Completeness \"customer_email\" >= 0.95, ColumnValues \"order_status\" in [\"PENDING\",\"SHIPPED\",\"DELIVERED\",\"RETURNED\"], ColumnValues \"net_amount\" between 0.0 and 25000.0 ]" \
    --target-table '{"TableName": "sales_data", "DatabaseName": "company_sales"}'

The --target-table values must match the ICEBERG_DATABASE and ICEBERG_TABLE environment variables used in the Pattern 2 job. The job discovers the ruleset at runtime using list_data_quality_rulesets filtered by that target table.

To validate data quality inline using PySpark

To implement this pattern, see the following code sample which can be run on all three Amazon EMR deployment modes:

import os
import boto3
from pyspark.sql import SparkSession
from awsdataquality.transforms import EvaluateDataQuality

spark = (SparkSession.builder.appName("blog-wap-dq")
    .config("spark.sql.catalog.iceberg", "org.apache.iceberg.spark.SparkCatalog")
    .config("spark.sql.catalog.iceberg.warehouse", "s3://amzn-s3-demo-bucket/warehouse/")
    .config("spark.sql.catalog.iceberg.catalog-impl", "org.apache.iceberg.aws.glue.GlueCatalog")
    .enableHiveSupport().getOrCreate())

database = os.environ["ICEBERG_DATABASE"]
table = os.environ["ICEBERG_TABLE"]
input_path = os.environ["WAP_INPUT_PATH"]
threshold = float(os.environ.get("WAP_SCORE_THRESHOLD", "1.0"))
region = os.environ.get("AWS_REGION", "us-east-1")
ruleset_name = os.environ.get("GLUE_DQ_RULESET_NAME")

# Branch name carries the Spark application id so concurrent nightly runs, and reruns
# of a failed one, never collide on the same branch.
branch = os.environ.get("WAP_BRANCH") or f"audit_{spark.sparkContext.applicationId}"
branch = branch.replace("-", "_")
fqtn = f"iceberg.`{database}`.`{table}`"
branch_fqtn = f"iceberg.`{database}`.`{table}`.branch_{branch}"

glue = boto3.client("glue", region_name=region)

# ---------------------------------------------------------- 1. DQ RULESET
# We will then get the data quality rules.
ruleset = glue.get_data_quality_ruleset(Name=ruleset_name)["Ruleset"]
print(f"auditing against catalog ruleset '{ruleset_name}':\n{ruleset}")

# ---------------------------------------------------------- 2. WRITE
main_before = spark.table(fqtn).count()
print(f"main has {main_before} row(s) before this batch")

# CREATE BRANCH (not CREATE OR REPLACE): if the branch somehow already exists, that is
# a concurrent or half-finished run and silently replacing it would discard its state.
spark.sql(f"ALTER TABLE {fqtn} CREATE BRANCH {branch}")
print(f"created branch {branch}")

# Write out our batch to our new branch
batch = spark.read.parquet(input_path)
batch.writeTo(branch_fqtn).append()
print(f"wrote {batch.count()} row(s) to branch {branch}")

# Reread the branch and create a data frame to perform our data quality checks.
branch_df = spark.read.option("branch", branch).table(f"iceberg.`{database}`.`{table}`")

# ---------------------------------------------------------- 3. AUDIT
# Will now evaluate the ruleset against the branch
publishing_opts = {
    "dataQualityEvaluationContext": ruleset_name,
    "publishToDataCatalogTarget": {
        "DatabaseName": database,
        "TableName": table
    },
    "enableDataQualityResultsPublishing": "true"
}

# See the AWS Glue Data Quality additional_options reference for advanced configuration:
# https://docs.aws.amazon.com/glue/latest/dg/data-quality-additional-options.html
additional_opts = {}
additional_sources = {}

# Run EvaluateDataQuality.
dq_results = EvaluateDataQuality.apply(branch_df, ruleset, publishing_opts,
    additional_opts, additional_sources)

# ---------------------------------------------------------- 4. PUBLISH or discard
# Iterate through the DQ results for any failed rules and if there are any failed
# rules, throw an exception to stop the job.
failures = []
for row in dq_results.collect():
    if row["Outcome"] == "Failed":
        failures.append(f"Rule: {row['Rule']} | Reason: {row['FailureReason']}")

# If there are any failures of any rule, we will drop the branch, and throw an error
if failures:
    # Drop first, then raise: an exception alone would leave the rejected branch (and
    # its data files) behind on the table for every failed night.
    spark.sql(f"ALTER TABLE {fqtn} DROP BRANCH {branch}")
    main_after = spark.table(fqtn).count()
    print(f"REJECTED: branch {branch} dropped; main unchanged at {main_after} row(s)")
    failure_summary = "\n".join(r for r in failures)
    raise RuntimeError(f"Data quality checks failed:\n{failure_summary}")

# fast_forward moves main's pointer to the branch's snapshot. It requires main to
# be an ancestor of the branch, which holds here because the branch was cut from
# main and only appended to --- no rewrite of the promoted data, no second copy.
spark.sql(f"CALL iceberg.system.fast_forward("
    f"table => '{database}.{table}', branch => 'main', to => '{branch}')")
spark.sql(f"ALTER TABLE {fqtn} DROP BRANCH {branch}")
main_after = spark.table(fqtn).count()
print(f"PUBLISHED: main fast-forwarded to {branch} "
    f"({main_before} -> {main_after} row(s)); branch dropped")

spark.stop()

This code demonstrates a sample WAP approach using Apache Iceberg. We start by retrieving the ruleset from the AWS Glue service that will run against the table. We then create a branch within the Apache Iceberg table and apply new data against it. We then run our data quality checks using EvaluateDataQuality with the ruleset that we earlier obtained. If any of the data quality rules fail, a branch that holds the new version of the table gets dropped. Otherwise, it will be committed as the new main branch and made available for all consumers.

In this example, you will notice that the publishing_opts (publishing options) look different. For the dataQualityEvaluationContext, we set the ruleset name. We also fill in the publishToDataCatalogTarget with the database and table name within the AWS Glue Data Catalog. This tags the data quality run and its results to that table for easier lookups if you want to view the history of data quality runs on the table.

Clean up

To avoid ongoing charges, remove the resources that you no longer need:

1. Shut down Amazon EMR on EC2 clusters, Amazon EMR Serverless applications, or Amazon EMR on EKS.
2. Delete the tables that you created. The table mentioned in this blog post is company_sales.sales_data.
3. Delete the data quality ruleset created in Pattern 2.
4. Delete any Amazon S3 data that was generated for the purposes of this blog post.

Conclusion

Data quality is a concern for most production data pipelines. Until now, teams running on Amazon EMR had to build that quality layer themselves or bring in AWS Glue as a separate processing engine to get managed quality capabilities.

With AWS Glue Data Quality now bundled in AWS runtime for Apache Spark (emr-spark-8.1.0) and later versions, that gap is closed. You write your DQDL rules once and store them in the AWS Glue Data Catalog. From there, run them across each of the three Amazon EMR deployment modes.

The patterns in this post represent two points on that spectrum: a lightweight inline check for ad hoc workloads, and a full WAP workflow for production pipelines that need atomic, auditable data promotion.

To learn more:

• AWS Glue Data Quality documentation – Full DQDL rule syntax and rule type reference.
• Amazon EMR Serverless documentation – Getting started with Amazon EMR Serverless.
• Apache Iceberg branching and Write-Audit-Publish – Full reference for the WAP pattern used in Pattern 2.
• Amazon EMR release 8.1 notes – Bundled libraries and changes in release 8.1.

Have questions or feedback on this post? Leave a comment below or reach out on the AWS re:Post community.