Getting started with Amazon EMR Data Quality

Create an Amazon EMR cluster or application. Use release emr-spark-8.1.0 or later on Amazon EMR on EC2, Amazon EMR on EKS, or Amazon EMR Serverless. Attach a role that has the permissions listed in IAM permissions.

For example, to create an Amazon EMR on EC2 cluster, run:

aws emr create-cluster --release-label emr-spark-8.1.0 \
--applications Name=Spark \
--region us-east-1 \
--name My_Spark_DataQuality_Cluster \
--log-uri s3://amzn-s3-demo-bucket/ \
--instance-type m5.xlarge \
--instance-count 2 \
--service-role EMR_DefaultRole_V2 \
--ec2-attributes InstanceProfile=EMR_EC2_DefaultRole,SubnetId=subnet-1234567890abcdef0

Define a ruleset and evaluate it. The following PySpark script defines a ruleset in DQDL, evaluates it against a DataFrame, and prints the per-rule outcome.

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

spark = SparkSession.builder.appName("dq-example").getOrCreate()

df = spark.read.parquet("s3://amzn-s3-demo-bucket/orders/")

ruleset = """
Rules = [
    RowCount > 1000,
    IsComplete "order_id",
    IsUnique "order_id",
    ColumnValues "status" in ["PLACED", "SHIPPED", "CANCELED"],
    Mean "order_total" between 10 and 500
]
"""

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

results.show(truncate=False)

The results DataFrame contains one row per rule with the following columns: Rule, Outcome, FailureReason, and EvaluatedMetrics. Rows with an Outcome of Failed include the observed value in FailureReason.

Submit the application. Submit the script the same way you submit any other Spark application for your Amazon EMR deployment mode.

Retrieve the result. After the application finishes, list the run and fetch its detail with the Amazon Glue Data Quality APIs or the Amazon CLI. If your application also sets publishToDataCatalogTarget with a DatabaseName and TableName, the run appears on the Data quality tab of that catalog table in the Amazon Glue console (see Viewing results in the Amazon Glue console). Otherwise, use the API or Amazon CLI to retrieve the run.

aws glue list-data-quality-results-v2 --max-results 5

aws glue get-data-quality-result-v2 --result-id <result-id>

To narrow the list to runs of a specific dataset, pass the EvaluationContext that your application used (for example, orders_daily) to ListDataQualityStatistics. When your application reads from a Amazon Glue Data Catalog table, you can additionally filter list-data-quality-results-v2 by DataSource.GlueTable.

Create the ruleset in Amazon Glue. Author and save the ruleset using one of the following paths:

Grant the Spark workload role access. Add glue:GetDataQualityRuleset to the role that your Amazon EMR Spark application assumes.

Fetch the ruleset in your Spark script. Call GetDataQualityRuleset with boto3, extract the Ruleset field, and pass the DQDL string to EvaluateDataQuality.apply.

import boto3
from awsdataquality.transforms import EvaluateDataQuality

glue = boto3.client("glue")
response = glue.get_data_quality_ruleset(Name="orders_quality")
ruleset = response["Ruleset"]

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