Amazon EMR Data Quality の開始方法

このセクションでは、Amazon EMR Data Quality の最初の評価をエンドツーエンドで実行する方法について説明します。Amazon EMR Spark アプリケーションに必要なアクセス許可と、Amazon EMR の DataFrame に対して DQDL ルールセットを送信して結果を取得するチュートリアルです。

• 前提条件

• 最初の評価を実行する

前提条件

IAM アクセス許可

Amazon EMR Spark ワークロードが引き受ける IAM ロール - クラスターにアタッチされた EC2 インスタンスプロファイル、またはステップごとのランタイムロール - は、Data Quality を実行して結果を公開するために次のアクセス許可を付与する必要があります。

アクション	必要な場合
glue:PublishDataQuality	各評価を Glue Data Quality AWS に発行して、履歴の表示、実行の比較、異常検出のフィードを行うことができます。発行はデフォルトでオン (enableDataQualityResultsPublishing は True) になっているため、 オプションを に設定して明示的にオプトアウトしない限り、このアクセス許可を付与しますFalse。
glue:AccessDataQualityRuntimeConfiguration	ライブラリが発行を初期化するときに、サービスから発行設定を取得するには。発行が有効になっているたびに必要です。有効にしないと、ライブラリは実行の発行をサイレントに無効にします。
cloudwatch:PutMetricData	ダッシュボードとアラームの実行ごとの合格と不合格のメトリクスを Amazon CloudWatch に出力するには。を enableDataQualityCloudWatchMetricsに設定した場合にのみ必要ですTrue。
glue:GetTable, glue:GetDatabase	カタログテーブルメタデータを読み取るには。入力 DataFrame が AWS Glue データカタログテーブルから読み取られるとき、またはサービスが実行を特定のカタログテーブルに属性publishToDataCatalogTarget付けできるようにアプリケーションが渡すときに必要です。
glue:GetDataQualityRuleset	AWS Glue データカタログから保存されたルールセットを取得するには。Spark スクリプトが DQDL をインラインで定義するのではなく、カタログからルールセットをロードする場合にのみ必要です。「AWS Glue データカタログから保存されたルールセットを取得する」を参照してください。

次のインラインポリシーは、Amazon EMR Spark アプリケーションから Data Quality の結果を発行するために必要な最小限のアクセス許可のセットを付与します。

{
    "Version": "2012-10-17",
    "Statement": [
        {
            "Sid": "GlueDataQuality",
            "Effect": "Allow",
            "Action": [
                "glue:PublishDataQuality",
                "glue:AccessDataQualityRuntimeConfiguration"
            ],
            "Resource": "*"
        }
    ]
}

最初の評価を実行する

1. Amazon EMR クラスターまたはアプリケーションを作成します。EC2 上の Amazon EMR、EKS 上の Amazon EMR、または Amazon EMR Serverless でリリース emr-spark-8.1.0以降を使用します。に記載されているアクセス許可を持つロールをアタッチしますIAM アクセス許可。

   たとえば、EC2 クラスターで Amazon EMR を作成するには、以下を実行します。

   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

2. ルールセットを定義して評価します。次の PySpark スクリプトは、DQDL のルールセットを定義し、DataFrame に対して評価して、ルールごとの結果を出力します。

   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)

   results DataFrame には、、Rule、Outcome、FailureReasonおよび の列を含むルールごとに 1 行が含まれますEvaluatedMetrics。が Outcome の行には、 の観測値Failedが含まれますFailureReason。

3. アプリケーションを送信します。Amazon EMR デプロイモードの他の Spark アプリケーションを送信するのと同じ方法でスクリプトを送信します。

   Amazon EMR on EC2

   aws emr add-steps --cluster-id j-1234567890ABC \
   --steps Type=Spark,Name="dq-example",ActionOnFailure=CONTINUE,Args=[s3://amzn-s3-demo-bucket/scripts/dq_example.py]

   Amazon EMR on EKS

   aws emr-containers start-job-run \
   --virtual-cluster-id <virtual-cluster-id> \
   --name dq-example \
   --execution-role-arn <execution-role-arn> \
   --release-label emr-spark-8.1.0 \
   --job-driver '{
       "sparkSubmitJobDriver": {
           "entryPoint": "s3://amzn-s3-demo-bucket/scripts/dq_example.py"
       }
   }'

   Amazon EMR Serverless

   aws emr-serverless start-job-run \
   --application-id <application-id> \
   --execution-role-arn <execution-role-arn> \
   --name dq-example \
   --job-driver '{
       "sparkSubmit": {
           "entryPoint": "s3://amzn-s3-demo-bucket/scripts/dq_example.py"
       }
   }'

4. 結果を取得します。アプリケーションが終了したら、実行を一覧表示し、 Glue Data Quality APIsまたは AWS を使用してその詳細を取得します AWS CLI。アプリケーションで publishToDataCatalogTargetDatabaseNameと も設定されている場合TableName、実行は Glue AWS コンソールのカタログテーブルのデータ品質タブに表示されます (「」を参照Glue AWS コンソールでの結果の表示)。それ以外の場合は、 API または AWS CLI を使用して実行を取得します。

   aws glue list-data-quality-results-v2 --max-results 5
   
   aws glue get-data-quality-result-v2 --result-id <result-id>

   特定のデータセットの実行にリストを絞り込むには、アプリケーションEvaluationContextが使用した ( などorders_daily) を に渡しますListDataQualityStatistics。アプリケーションが AWS Glue データカタログテーブルから読み取ると、さらに list-data-quality-results-v2でフィルタリングできますDataSource.GlueTable。

AWS Glue データカタログから保存されたルールセットを取得する

各 Spark スクリプトでルールセットをインラインで定義する代わりに、ルールセットを AWS Glue データカタログに一度作成し、実行時に Amazon EMR Spark アプリケーションから取得できます。これにより、ルールセットを 1 か所でバージョン化し、Amazon EMR Spark ワークロード、 AWS Glue ETL ジョブ、評価間で共有できます。

1. Glue AWS でルールセットを作成します。次のいずれかのパスを使用してルールセットを作成して保存します。

   • Glue AWS コンソールで。「 Glue デベロッパーガイド」の「新しいルールセットの作成」を参照してください。 AWS

   • Glue Data Quality API AWS 経由。AWS Glue API リファレンスのCreateDataQualityRuleset「」を参照してください。

2. Spark ワークロードロールにアクセス権を付与します。Amazon EMR Spark アプリケーションが引き受けるロールglue:GetDataQualityRulesetに を追加します。

3. Spark スクリプトでルールセットを取得します。boto3 GetDataQualityRulesetで を呼び出し、 Rulesetフィールドを抽出し、DQDL 文字列を に渡します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,
       },
   )