Services or capabilities described in Amazon Web Services documentation might vary by Region. To see the differences applicable to the China Regions, see Getting Started with Amazon Web Services in China (PDF).

Amazon EMR Data Quality

Amazon EMR Data Quality, powered by Amazon Glue Data Quality, lets you measure and monitor the quality of the data that your Amazon EMR Spark workloads read and write without leaving the Spark application. Amazon EMR Data Quality uses the same open-source Deequ engine as Amazon Glue Data Quality. Starting with the AWS runtime for Apache Spark (emr-spark-8.1.0) on Amazon EMR, you can evaluate data quality directly inside your Amazon EMR Spark applications by adding the EvaluateDataQuality transform to your Spark script. The transform runs a declarative ruleset against your DataFrame and returns the per-rule outcome in-process, so your application can act on the result (for example, aborting a downstream write when a rule fails). Each evaluation is also published to Amazon Glue Data Quality, so you can review the history of your runs and alert on them alongside runs from Amazon Glue ETL and the Amazon Glue Data Catalog.

Using Amazon EMR Data Quality requires a Spark application and a ruleset written in Data Quality Definition Language (DQDL). No other programming is required beyond your existing Spark workflow.

Amazon EMR Data Quality is available on all three Amazon EMR deployment modes: Amazon EMR on EC2, Amazon EMR on EKS, and Amazon EMR Serverless. It is supported for Apache Spark only.

How Data Quality works

Amazon EMR Data Quality is a Spark transform. You pass it a DataFrame and a DQDL ruleset; it evaluates the rules and returns a DataFrame of outcomes. Each evaluation is also published to Amazon Glue Data Quality so you can query the result later through the APIs or the Amazon Glue console. Amazon Glue ETL and the Amazon Glue Data Catalog publish to the same store, so you can compare Amazon EMR runs with runs from either of those engines.

You define your data quality expectations as a declarative ruleset written in Data Quality Definition Language (DQDL) reference. A ruleset is a set of rules that describe expected properties of a dataset. For example, the following ruleset checks that a table has more than 100 rows, that its customer_id column is complete and unique, and that the country column is drawn from a fixed set of values.

Rules = [
    RowCount > 100,
    IsComplete "customer_id",
    IsUnique "customer_id",
    ColumnValues "country" in ["US", "CA", "MX"]
]

At runtime, the EvaluateDataQuality transform reads your DataFrame, evaluates each rule with the same Deequ engine used by Amazon Glue Data Quality and the Amazon Glue Data Catalog, and returns the outcome (Passed, Failed, or Error) for every rule. Failed rules include the observed value so you can triage them without re-running the application. The result is a Spark DataFrame your application can branch on, log, or persist.

Data Quality also publishes each evaluation to the Amazon Glue Data Quality service. From there you can:

Because results can be published to the same store used by Amazon Glue Data Quality, you can correlate runs across compute engines. Runs that share an EvaluationContext (a string identifier that you pass to the transform) appear as a single logical dataset in the console and in anomaly-detection training, regardless of whether they came from an Amazon EMR Spark application or a Amazon Glue ETL job.

Concepts and terminology

The following terms appear throughout this chapter and in the Amazon Glue Data Quality APIs and console.

Where to validate in your pipeline

You can invoke EvaluateDataQuality anywhere in a Spark script. The three common patterns are:

The following PySpark snippet gates a write on the result of a post-validation evaluation.

results = EvaluateDataQuality.apply(
    frame=final_df,
    ruleset=ruleset,
    publishing_options={
        "dataQualityEvaluationContext": "orders_daily",
        "enableDataQualityResultsPublishing": True,
    },
)

failed = results.filter("Outcome in ('Failed', 'Error')").count()
if failed > 0:
    raise RuntimeError(f"{failed} data quality rules failed or errored; aborting write")

final_df.write.mode("overwrite").parquet("s3://amzn-s3-demo-bucket/orders_curated/")

Limits