# アラート、デプロイ、スケジュールの設定 このトピックでは、AWS Glue Data Quality の、アラート、デプロイ、スケジュールの設定方法について説明します。 **Contents** + [Amazon EventBridge 統合でのアラートと通知の設定](#data-quality-alerts-eventbridge) + [イベントパターンの追加設定のオプション](#data-quality-alerts-eventbridge-config-options) + [通知をメールとしてフォーマットする](#data-quality-alerts-eventbridge-format-notifications) + [CloudWatch 統合でアラートと通知をセットアップする](#data-quality-alerts-cloudwatch) + [データ品質評価の結果をクエリしてダッシュボードを作成する](#data-quality-alerts-querying-results) + [AWS CloudFormation を使用したデータ品質ルールのデプロイ](#data-quality-deploy-cfn) + [データ品質ルールのスケジューリング](#data-quality-scheduling-rules) ## Amazon EventBridge 統合でのアラートと通知の設定 AWS Glue Data Quality は EventBridge イベントのパブリッシュに対応しています。このイベントは、Data Quality ルールセットの評価実行が完了すると生成されます。これにより、Data Quality のルールが失敗した場合のアラートを、簡単に設定できます。 以下は、データカタログの Data Quality ルールセットを評価する際の、イベントのサンプルです。この情報を使用すれば、Amazon EventBridge で公開されているデータを確認できます。追加の API 呼び出しを発行すると、さらに詳細を入手できます。例えば、特定の実行の詳細を取得するには、結果 ID を使用して `get_data_quality_result` API を呼び出します。 ``` { "version":"0", "id":"abcdef00-1234-5678-9abc-def012345678", "detail-type":"Data Quality Evaluation Results Available", "source":"aws.glue-dataquality", "account":"123456789012", "time":"2017-09-07T18:57:21Z", "region":"us-west-2", "resources":[], "detail":{ "context": { "contextType": "GLUE_DATA_CATALOG", "runId":"dqrun-12334567890", "databaseName": "db-123", "tableName": "table-123", "catalogId": "123456789012" }, "resultID": "dqresult-12334567890", "rulesetNames": ["rulset1"], "state":"SUCCEEDED", "score": 1.00, "rulesSucceeded": 100, "rulesFailed": 0, "rulesSkipped": 0 } } ``` こちらは、AWS Glue ETL または AWS Glue Studio ノートブックで Data Quality ルールセットを評価するときにパブリッシュする、サンプルのイベントです。 ``` { "version":"0", "id":"abcdef00-1234-5678-9abc-def012345678", "detail-type":"Data Quality Evaluation Results Available", "source":"aws.glue-dataquality", "account":"123456789012", "time":"2017-09-07T18:57:21Z", "region":"us-west-2", "resources":[], "detail":{ "context": { "contextType": "GLUE_JOB", "jobId": "jr-12334567890", "jobName": "dq-eval-job-1234", "evaluationContext": "", } "resultID": "dqresult-12334567890", "rulesetNames": ["rulset1"], "state":"SUCCEEDED", "score": 1.00 "rulesSucceeded": 100, "rulesFailed": 0, "rulesSkipped": 0 } } ``` データカタログと ETL ジョブの両方で Data Quality 評価を実行する場合は、EventBridge のパブリッシュを実行するために、デフォルトで選択されている **[Publish metrics to Amazon CloudWatch]** を選択したままにしておきます。 **EventBridge 通知のセットアップ** ![\[AWS CloudFormation のデータ品質プロパティ\]](http://docs.aws.amazon.com/ja_jp/glue/latest/dg/images/data-quality-properties-cfn.png) 生成されたイベントを受信してターゲットを定義するには、Amazon EventBridge ルールを設定する必要があります。ルールを作成するには 1. Amazon EventBridge コンソールを開きます。 1. ナビゲーションバーの **[バス]** セクションで **[ルール]** を選択します。 1. **[Create Rule]** (ルールの作成) を選択します。 1. **[ルールの詳細を定義]** では以下のとおり入力および選択します。 1. [名前] に `myDQRule` と入力します。 1. 説明を記入します (任意)。 1. [イベントバス] では、お使いのイベントバスを選択します。お持ちでない場合は、デフォルトのままにしておきます。 1. [ルールタイプ] では **[イベントパターンを持つルール]** を選択し、続いて **[次へ]** を選択します。 1. **[イベントパターンを構築]** では以下のとおり入力および選択します。 1. [イベントソース] では **[AWS イベントまたは EventBridge パートナーイベント]** を選択します。 1. [サンプルイベント] セクションは飛ばします。 1. [作成のメソッド] では **[パターンフォームを使用する]** を選択します。 1. イベントパターンの場合 1. イベントソースに **[AWS のサービス]** を選択します。 1. AWS サービスに **[Glue Data Quality]** を選択します。 1. イベントタイプに **[Data Quality Evaluation Results Available]** を選択します。 1. 具体的な状態に **[FAILED]** を選択します。そうすると、次のようなイベントパターンが表示されます。 ``` { "source": ["aws.glue-dataquality"], "detail-type": ["Data Quality Evaluation Results Available"], "detail": { "state": ["FAILED"] } } ``` 1. 設定オプションの詳細については、「[イベントパターンの追加設定のオプション](#data-quality-alerts-eventbridge-config-options)」を参照してください。 1. **[ターゲットを選択]** では以下のとおり入力および選択します。 1. **[ターゲットタイプ]** で **[AWS のサービス]** を選択します。 1. **[ターゲットを選択]** ドロップダウンで、(SNS、Lambda、SQS などへの) 接続に使用する AWS サービスを選択して、**[次へ]** を選択します。 1. **[Configure tag(s)]** で、**[Add new tags]** をクリックし、オプションのタグを追加して、**[次へ]** を選択します。 1. すべての選択を示す概要ページが表示されます。一番下で **[ルールの作成]** を選択します。 ### イベントパターンの追加設定のオプション イベントを、成功または失敗に応じてフィルタリングするだけでなく、さまざまなパラメータに応じてさらにフィルタリングするとよい場合があります。 これを行うには、[イベントパターン] のセクションに進み、**[パターンを編集]** を選択して追加のパラメータを指定します。イベントパターンのフィールドは大文字と小文字を区別する必要があります。以下は、イベントパターンの設定の例です。 特定のルールセットを評価する特定のテーブルからイベントを取り込むときは、次のタイプのパターンを使用します。 ``` { "source": ["aws.glue-dataquality"], "detail-type": ["Data Quality Evaluation Results Available"], "detail": { "context": { "contextType": ["GLUE_DATA_CATALOG"], "databaseName": "db-123", "tableName": "table-123", }, "rulesetNames": ["ruleset1", "ruleset2"] "state": ["FAILED"] } } ``` ETL 体験の特定のジョブからイベントを取り込むときは、次のタイプのパターンを使用します。 ``` { "source": ["aws.glue-dataquality"], "detail-type": ["Data Quality Evaluation Results Available"], "detail": { "context": { "contextType": ["GLUE_JOB"], "jobName": ["dq_evaluation_job1", "dq_evaluation_job2"] }, "state": ["FAILED"] } } ``` スコアが特定のしきい値 (70% など) を下回るイベントを取り込むときは、次のタイプのパターンを使用します。 ``` { "source": ["aws.glue-dataquality"], "detail-type": ["Data Quality Evaluation Results Available"], "detail": { "score": [{ "numeric": ["<=", 0.7] }] } } ``` ### 通知をメールとしてフォーマットする 自社のビジネスチームに、適切にフォーマットしたメール通知を送信することが必要になる場合があります。Amazon EventBridge と AWS を使用すればこれを実行できます。 ![\[メールにフォーマットされたデータ品質通知\]](http://docs.aws.amazon.com/ja_jp/glue/latest/dg/images/data_quality_sample_email.png) 次のサンプルコードを使用すると、データ品質通知をフォーマットしてメールを生成できます。 ``` import boto3 import json from datetime import datetime sns_client = boto3.client('sns') glue_client = boto3.client('glue') sns_topic_arn = 'arn:aws:sns:::' def lambda_handler(event, context): log_metadata = {} message_text = "" subject_text = "" if event['detail']['context']['contextType'] == 'GLUE_DATA_CATALOG': log_metadata['ruleset_name'] = str(event['detail']['rulesetNames'][0]) log_metadata['tableName'] = str(event['detail']['context']['tableName']) log_metadata['databaseName'] = str(event['detail']['context']['databaseName']) log_metadata['runId'] = str(event['detail']['context']['runId']) log_metadata['resultId'] = str(event['detail']['resultId']) log_metadata['state'] = str(event['detail']['state']) log_metadata['score'] = str(event['detail']['score']) log_metadata['numRulesSucceeded'] = str(event['detail']['numRulesSucceeded']) log_metadata['numRulesFailed'] = str(event['detail']['numRulesFailed']) log_metadata['numRulesSkipped'] = str(event['detail']['numRulesSkipped']) message_text += "Glue Data Quality run details:\n" message_text += "ruleset_name: {}\n".format(log_metadata['ruleset_name']) message_text += "glue_table_name: {}\n".format(log_metadata['tableName']) message_text += "glue_database_name: {}\n".format(log_metadata['databaseName']) message_text += "run_id: {}\n".format(log_metadata['runId']) message_text += "result_id: {}\n".format(log_metadata['resultId']) message_text += "state: {}\n".format(log_metadata['state']) message_text += "score: {}\n".format(log_metadata['score']) message_text += "numRulesSucceeded: {}\n".format(log_metadata['numRulesSucceeded']) message_text += "numRulesFailed: {}\n".format(log_metadata['numRulesFailed']) message_text += "numRulesSkipped: {}\n".format(log_metadata['numRulesSkipped']) subject_text = "Glue Data Quality ruleset {} run details".format(log_metadata['ruleset_name']) else: log_metadata['ruleset_name'] = str(event['detail']['rulesetNames'][0]) log_metadata['jobName'] = str(event['detail']['context']['jobName']) log_metadata['jobId'] = str(event['detail']['context']['jobId']) log_metadata['resultId'] = str(event['detail']['resultId']) log_metadata['state'] = str(event['detail']['state']) log_metadata['score'] = str(event['detail']['score']) log_metadata['numRulesSucceeded'] = str(event['detail']['numRulesSucceeded']) log_metadata['numRulesFailed'] = str(event['detail']['numRulesFailed']) log_metadata['numRulesSkipped'] = str(event['detail']['numRulesSkipped']) message_text += "Glue Data Quality run details:\n" message_text += "ruleset_name: {}\n".format(log_metadata['ruleset_name']) message_text += "glue_job_name: {}\n".format(log_metadata['jobName']) message_text += "job_id: {}\n".format(log_metadata['jobId']) message_text += "result_id: {}\n".format(log_metadata['resultId']) message_text += "state: {}\n".format(log_metadata['state']) message_text += "score: {}\n".format(log_metadata['score']) message_text += "numRulesSucceeded: {}\n".format(log_metadata['numRulesSucceeded']) message_text += "numRulesFailed: {}\n".format(log_metadata['numRulesFailed']) message_text += "numRulesSkipped: {}\n".format(log_metadata['numRulesSkipped']) subject_text = "Glue Data Quality ruleset {} run details".format(log_metadata['ruleset_name']) resultID = str(event['detail']['resultId']) response = glue_client.get_data_quality_result(ResultId=resultID) RuleResults = response['RuleResults'] message_text += "\n\nruleset details evaluation steps results:\n\n" subresult_info = [] for dic in RuleResults: subresult = "Name: {}\t\tResult: {}\t\tDescription: \t{}".format(dic['Name'], dic['Result'], dic['Description']) if 'EvaluationMessage' in dic: subresult += "\t\tEvaluationMessage: {}".format(dic['EvaluationMessage']) subresult_info.append({ 'Name': dic['Name'], 'Result': dic['Result'], 'Description': dic['Description'], 'EvaluationMessage': dic.get('EvaluationMessage', '') }) message_text += "\n" + subresult log_metadata['resultrun'] = subresult_info sns_client.publish( TopicArn=sns_topic_arn, Message=message_text, Subject=subject_text ) return { 'statusCode': 200, 'body': json.dumps('Message published to SNS topic') } ``` ## CloudWatch 統合でアラートと通知をセットアップする 当社では、Amazon EventBridge を使用してデータ品質アラートをセットアップする方法を推奨しています。Amazon EventBridge ならアラートのセットアップが 1 回で済むためです。ただし使い慣れているからという理由で Amazon CloudWatch を選ぶユーザーもいます。このようなユーザーのために当社は Amazon CloudWatch との統合をご用意しています。 各 AWS Glue Data Quality 評価は、データ品質の実行のたびに、`glue.data.quality.rules.passed` (合格したルールの数を示す) と `glue.data.quality.rules.failed` (不合格だったルールの数を示す) から成る 1 組のメトリクスを発行します。所定のデータ品質の実行がしきい値を下回った場合、発行されたこのメトリクスを使用してアラームを作成し、ユーザーに警告できます。Amazon SNS 通知からメールを送信するアラームをセットアップするには、以下の手順に従います。 Amazon SNS 通知からメールを送信するアラームをセットアップするには、以下の手順に従います。 1. Amazon CloudWatch コンソールを開きます。 1. **[メトリクス]** で、**[すべてのメトリクス]** を選択します。[カスタム名前空間] に、Glue Data Quality というタイトルの追加の名前空間が表示されています。 **注記** AWS Glue Data Quality の実行を始めるときは、**[Publish metrics to Amazon CloudWatch]** のチェックボックスがオンになっていることを確認します。オンになっていないと、この実行用のメトリクスが Amazon CloudWatch にパブリッシュされません。 `Glue Data Quality` 名前空間で、テーブルごと、ルールセットごとに発行されるメトリクスを確認できます。このトピックでは `glue.data.quality.rules.failed` ルールを使用し、この値が 1 を超えた場合にアラームを発行する (つまり、不合格と評価されたルールの数が 1 を超えた場合に通知する) ようにします。 1. アラームを作成するには、**[アラーム]** で、**[すべてのアラーム]** を選択します。 1. [**アラームの作成**] を選択します。 1. **メトリクスの選択** を選択します。 1. 作成したテーブルに対応する `glue.data.quality.rules.failed` メトリクスを選択し、次に **[メトリクスの選択]** を選択します。 1. **[メトリクスと条件の指定]** タブの **[メトリクス]** セクションで、以下のとおり選択します。 1. **[統計]** で、**[合計]** を選択します。 1. **[期間]** は、**[1 minute]** を選択します。 1. **[条件]** セクションで、以下のとおり選択および入力します。 1. [**Threshold type**] で [**静的**] を選択します。 1. **[Whenever glue.data.quality.rules.failed is...**] では、**[以上]** を選択します。 1. **[than...]** に、しきい値として **1** と入力します。 上記を選択したことにより、`glue.data.quality.rules.failed` メトリクスで 1 以上の値が発行されると、アラームがトリガーされることになります。ただし、データがない場合は許容範囲として処理されます。 1. [**次へ**] を選択します。 1. **[アクションの設定]** で、以下のとおり選択および入力します。 1. **[アラーム状態トリガー]** セクションで、**[アラーム状態]** を選択します。 1. **[Send a notification to the following SNS topic]** セクションで、**[Create a new topic to send a notification via a new SNS topic]** を選択します。 1. **[Email endpoints that will receive the notification]** に、自身のメールアドレスを入力します。**[トピックの作成]** をクリックします。 1. [**次へ**] を選択します。 1. **[アラーム名]** に `myFirstDQAlarm` と入力し、**[次へ]** を選択します。 1. すべての選択を示す概要ページが表示されます。一番下で **[アラームを作成]** を選択します。 作成したアラームは、Amazon CloudWatch アラームダッシュボードで確認できます。 ## データ品質評価の結果をクエリしてダッシュボードを作成する データ品質評価の結果を表示する、ダッシュボードの作成が必要になる場合があります。これには、以下の 2 つの方法があります。 **以下のコードを使用して、データを Amazon S3 に書き込むように Amazon EventBridge をセットアップする** ``` import boto3 import json from datetime import datetime s3_client = boto3.client('s3') glue_client = boto3.client('glue') s3_bucket = 's3-bucket-name' def write_logs(log_metadata): try: filename = datetime.now().strftime("%m%d%Y%H%M%S") + ".json" key_opts = { 'year': datetime.now().year, 'month': "{:02d}".format(datetime.now().month), 'day': "{:02d}".format(datetime.now().day), 'filename': filename } s3key = "gluedataqualitylogs/year={year}/month={month}/day={day}/{filename}".format(**key_opts) s3_client.put_object(Bucket=s3_bucket, Key=s3key, Body=json.dumps(log_metadata)) except Exception as e: print(f'Error writing logs to S3: {e}') def lambda_handler(event, context): log_metadata = {} message_text = "" subject_text = "" if event['detail']['context']['contextType'] == 'GLUE_DATA_CATALOG': log_metadata['ruleset_name'] = str(event['detail']['rulesetNames'][0]) log_metadata['tableName'] = str(event['detail']['context']['tableName']) log_metadata['databaseName'] = str(event['detail']['context']['databaseName']) log_metadata['runId'] = str(event['detail']['context']['runId']) log_metadata['resultId'] = str(event['detail']['resultId']) log_metadata['state'] = str(event['detail']['state']) log_metadata['score'] = str(event['detail']['score']) log_metadata['numRulesSucceeded'] = str(event['detail']['numRulesSucceeded']) log_metadata['numRulesFailed'] = str(event['detail']['numRulesFailed']) log_metadata['numRulesSkipped'] = str(event['detail']['numRulesSkipped']) message_text += "Glue Data Quality run details:\n" message_text += "ruleset_name: {}\n".format(log_metadata['ruleset_name']) message_text += "glue_table_name: {}\n".format(log_metadata['tableName']) message_text += "glue_database_name: {}\n".format(log_metadata['databaseName']) message_text += "run_id: {}\n".format(log_metadata['runId']) message_text += "result_id: {}\n".format(log_metadata['resultId']) message_text += "state: {}\n".format(log_metadata['state']) message_text += "score: {}\n".format(log_metadata['score']) message_text += "numRulesSucceeded: {}\n".format(log_metadata['numRulesSucceeded']) message_text += "numRulesFailed: {}\n".format(log_metadata['numRulesFailed']) message_text += "numRulesSkipped: {}\n".format(log_metadata['numRulesSkipped']) subject_text = "Glue Data Quality ruleset {} run details".format(log_metadata['ruleset_name']) else: log_metadata['ruleset_name'] = str(event['detail']['rulesetNames'][0]) log_metadata['jobName'] = str(event['detail']['context']['jobName']) log_metadata['jobId'] = str(event['detail']['context']['jobId']) log_metadata['resultId'] = str(event['detail']['resultId']) log_metadata['state'] = str(event['detail']['state']) log_metadata['score'] = str(event['detail']['score']) log_metadata['numRulesSucceeded'] = str(event['detail']['numRulesSucceeded']) log_metadata['numRulesFailed'] = str(event['detail']['numRulesFailed']) log_metadata['numRulesSkipped'] = str(event['detail']['numRulesSkipped']) message_text += "Glue Data Quality run details:\n" message_text += "ruleset_name: {}\n".format(log_metadata['ruleset_name']) message_text += "glue_job_name: {}\n".format(log_metadata['jobName']) message_text += "job_id: {}\n".format(log_metadata['jobId']) message_text += "result_id: {}\n".format(log_metadata['resultId']) message_text += "state: {}\n".format(log_metadata['state']) message_text += "score: {}\n".format(log_metadata['score']) message_text += "numRulesSucceeded: {}\n".format(log_metadata['numRulesSucceeded']) message_text += "numRulesFailed: {}\n".format(log_metadata['numRulesFailed']) message_text += "numRulesSkipped: {}\n".format(log_metadata['numRulesSkipped']) subject_text = "Glue Data Quality ruleset {} run details".format(log_metadata['ruleset_name']) resultID = str(event['detail']['resultId']) response = glue_client.get_data_quality_result(ResultId=resultID) RuleResults = response['RuleResults'] message_text += "\n\nruleset details evaluation steps results:\n\n" subresult_info = [] for dic in RuleResults: subresult = "Name: {}\t\tResult: {}\t\tDescription: \t{}".format(dic['Name'], dic['Result'], dic['Description']) if 'EvaluationMessage' in dic: subresult += "\t\tEvaluationMessage: {}".format(dic['EvaluationMessage']) subresult_info.append({ 'Name': dic['Name'], 'Result': dic['Result'], 'Description': dic['Description'], 'EvaluationMessage': dic.get('EvaluationMessage', '') }) message_text += "\n" + subresult log_metadata['resultrun'] = subresult_info write_logs(log_metadata) return { 'statusCode': 200, 'body': json.dumps('Message published to SNS topic') } ``` データを Amazon S3 に書き込むと、AWS Glue クローラーを使用して Athena に登録し、テーブルをクエリできます。 **データ品質評価中に Amazon S3 のロケーションを設定する** AWS Glue データカタログまたは AWS Glue ETL でデータ品質タスクを実行しているときは、Amazon S3 のロケーションを指定してデータ品質評価の結果を Amazon S3 に書き込むことが可能です。以下の構文を使用すると、ターゲットを参照してテーブルを作成し、データ品質評価の結果を読み取れます。 `CREATE EXTERNAL TABLE` と `MSCK REPAIR TABLE` のクエリは個別に実行する必要がありますので注意してください。 ``` CREATE EXTERNAL TABLE ( catalogid string, databasename string, tablename string, dqrunid string, evaluationstartedon timestamp, evaluationcompletedon timestamp, rule string, outcome string, failurereason string, evaluatedmetrics string) PARTITIONED BY ( `year` string, `month` string, `day` string) ROW FORMAT SERDE 'org.openx.data.jsonserde.JsonSerDe' WITH SERDEPROPERTIES ( 'paths'='catalogId,databaseName,dqRunId,evaluatedMetrics,evaluationCompletedOn,evaluationStartedOn,failureReason,outcome,rule,tableName') STORED AS INPUTFORMAT 'org.apache.hadoop.mapred.TextInputFormat' OUTPUTFORMAT 'org.apache.hadoop.hive.ql.io.HiveIgnoreKeyTextOutputFormat' LOCATION 's3://glue-s3-dq-bucket-us-east-2-results/' TBLPROPERTIES ( 'classification'='json', 'compressionType'='none', 'typeOfData'='file'); ``` ``` MSCK REPAIR TABLE ; ``` 上記のテーブルを作成すれば、Amazon Athena を使用して分析クエリを実行できるようになります。 ## AWS CloudFormation を使用したデータ品質ルールのデプロイ データ品質ルールの作成には、AWS CloudFormation を使用します。詳細については、「[AWS Glue 用の AWS CloudFormation](https://docs.aws.amazon.com/glue/latest/dg/populate-with-cloudformation-templates.html)」を参照してください。 ## データ品質ルールのスケジューリング データ品質ルールのスケジューリングには、次の方法を使用します。 + データカタログからデータ品質ルールをスケジュールします。ノーコードのユーザーは、この方法を使用することでデータ品質スキャンを簡単にスケジューリングできます。AWSGlue Data Quality では、Amazon EventBridge にスケジュールを作成します。データ品質ルールをスケジュールするには、以下の手順に従います。 + ルールセットに進み、**[実行]** をクリックします。 + **[Run frequency]** で希望するスケジュールを選択し、**タスク名**を入力します。このタスク名は、EventBridge でのスケジュールの名前です。 + Amazon EventBridge と AWS Step Functions を使用して、データ品質ルールの評価と推奨事項をオーケストレートします。