本文為英文版的機器翻譯版本，如內容有任何歧義或不一致之處，概以英文版為準。

# 設定提醒、部署和排程
<a name="data-quality-alerts"></a>

本主題說明如何設定 Glue Data Quality AWS 的提醒、部署和排程。

**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 整合中設定提醒和通知
<a name="data-quality-alerts-eventbridge"></a>

AWS Glue Data Quality 支援發佈 EventBridge 事件，這些事件會在 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
    }
}
```

以下是當您在 Glue ETL 或 AWS Glue Studio AWS 筆記本中評估資料品質規則集時發佈的範例事件。

```
{
    "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
    }
}
```

若要在 Data Catalog 和 ETL 任務中同時執行資料品質評估，**將指標發布至 Amazon CloudWatch** 選項 (預設已選取) 必須保持選取狀態，EventBridge 發布才能運作。

**設定 EventBridge 通知**

![\[AWS CloudFormation 中的資料品質屬性\]](http://docs.aws.amazon.com/zh_tw/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. 對於事件類型，選取**可用的資料品質評估結果**。

      1. 對於特定狀態，選取**失敗**。然後您會看到類似以下內容的事件模式：

         ```
         {
           "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. 使用**選取目標**下拉式清單，選擇要連線的所需 AWS 服務 (SNS、Lambda、SQS 等），然後選擇**下一步**。

1. 在**設定標籤**上按一下**新增標籤**以新增選用標籤，然後選擇**下一步**。

1. 您會看到所有選取項目的摘要頁面。選擇底部的**建立規則**。

### 事件模式的其他組態選項
<a name="data-quality-alerts-eventbridge-config-options"></a>

除了根據成功或失敗篩選事件之外，您可能還想要根據不同參數進一步篩選事件。

若要這麼做，請前往「事件模式」區段，然後選取**編輯模式**以指定其他參數。請注意，事件模式中的欄位需區分大小寫。以下是設定事件模式的範例。

若要從評估特定規則集的特定資料表擷取事件，請使用此類型的模式：

```
{
  "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]
    }]
  }
}
```

### 將通知格式化為電子郵件
<a name="data-quality-alerts-eventbridge-format-notifications"></a>

您有時需要向業務團隊傳送格式良好的電子郵件通知。您可以使用 Amazon EventBridge 和 AWS Lambda 來達成此目標。

![\[格式化為電子郵件的資料品質通知\]](http://docs.aws.amazon.com/zh_tw/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:<region-code>:<account-id>:<sns-topic-name>'



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 整合中設定提醒和通知
<a name="data-quality-alerts-cloudwatch"></a>

我們建議使用 Amazon EventBridge 設定資料品質提醒，因為 Amazon EventBridge 需要一次性設定來提醒客戶。但是，某些客戶偏好 Amazon CloudWatch，因為他們更熟悉此服務。對於這類客戶，我們提供與 Amazon CloudWatch 的整合。

每個 AWS Glue Data Quality 評估都會針對每個資料品質執行發出一對名為 `glue.data.quality.rules.passed`（表示通過的規則數） 和 `glue.data.quality.rules.failed`（表示失敗的規則數） 的指標。您可以使用發出的此指標來建立警示，以在指定的資料品質執行低於閾值時提醒使用者。若要開始設定將透過 Amazon SNS 通知傳送電子郵件的警示，請依照以下步驟操作：

若要開始設定將透過 Amazon SNS 通知傳送電子郵件的警示，請依照以下步驟操作：

1. 開啟 Amazon CloudWatch 主控台。

1. 選擇**指標**下的**所有指標**。您將在標題為 "Glue Data Quality" 的自訂命名空間下看到額外的命名空間。
**注意**  
啟動 AWS Glue Data Quality 執行時，請確定已啟用**發佈指標至 Amazon CloudWatch** 核取方塊。否則，該特定執行的指標將不會發布至 Amazon CloudWatch。

   在 `Glue Data Quality` 命名空間下，您可以看到每個資料表中依規則集發出的指標。本主題的目的在於，如果此值超過 1 (表示如果我們看到失敗規則評估的數量大於 1，我們希望收到通知)，我們將使用 `glue.data.quality.rules.failed` 規則和警示。

1. 若要建立警示，請選擇**警示**下的**所有警示**。

1. 選擇 **Create alarm** (建立警示)。

1. 選擇 **Select metric** (選取指標)。

1. 選取與您建立之資料表對應的 `glue.data.quality.rules.failed` 指標，然後選擇**選取指標**。

1. 在**指標**區段下的**指定指標和條件**索引標籤下：

   1. 在 **Statistic (統計資料)** 中選擇 **Sum (總和)**。

   1. 對於**期間**，選擇 **1 分鐘**。

1. 在**條件**區段下：

   1. 對於**閾值類型**，選擇**靜態**。

   1. 對於**每當 glue.data.quality.rules.failed 為…**，選取**大於/等於**。

   1. 對於**比…**，輸入 **1** 作為閾值。

   這些選擇意味著，如果 `glue.data.quality.rules.failed` 指標發出的值大於或等於 1，我們將觸發警示。但是，如果沒有資料，我們會將其視為可接受。

1. 選擇**下一步**。

1. 在**設定動作**上：

   1. 對於**警示狀態觸發**區段，選擇**警示中**。

   1. 對於**將通知傳送至下列 SNS 主題**區段，選擇**建立新主題以透過新的 SNS 主題傳送通知**。

   1. 對於**將接收通知的電子郵件端點**，輸入電子郵件地址。然後按一下**建立主題**。

   1. 選擇**下一步**。

1. 對於**警示名稱**，輸入 `myFirstDQAlarm`，然後選擇**下一步**。

1. 您會看到所有選取項目的摘要頁面。選擇底部的**建立警示**。

您現在可以從 Amazon CloudWatch 警示儀表板看到正在建立的警示。

## 查詢資料品質結果以建置儀表板
<a name="data-quality-alerts-querying-results"></a>

您可能想要建置儀表板以顯示資料品質結果。有兩種方式可以進行：

**使用下列程式碼設定 Amazon EventBridge，以將資料寫入 Amazon S3：**

```
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 Data Catalog 或 Glue ETL AWS 中執行資料品質任務時，您可以提供 Amazon S3 位置，將資料品質結果寫入 Amazon S3。您可以使用以下語法，透過參考目標以讀取資料品質結果來建立資料表。

請注意，您必須分別執行 `CREATE EXTERNAL TABLE` 和 `MSCK REPAIR TABLE` 查詢。

```
CREATE EXTERNAL TABLE <my_table_name>(
    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 <my_table_name>;
```

建立上述資料表後，您便可以使用 Amazon Athena 執行分析查詢。

## 使用 AWS CloudFormation 部署資料品質規則
<a name="data-quality-deploy-cfn"></a>

您可以使用 AWS CloudFormation 來建立資料品質規則。如需詳細資訊，請參閱 [AWS CloudFormation for AWS Glue](https://docs.aws.amazon.com/glue/latest/dg/populate-with-cloudformation-templates.html)。

## 排程資料品質規則
<a name="data-quality-scheduling-rules"></a>

您可以使用下列方法排程資料品質規則：
+  從 Data Catalog 排程資料品質規則：沒有程式碼使用者可以使用此選項來輕鬆排程其資料品質掃描。 AWS Glue Data Quality 將在 Amazon EventBridge 中建立排程。若要排程資料品質規則：
  +  導覽至規則集，然後按一下**執行**。
  +  在**執行頻率**中，選取所需排程並提供**任務名稱**。此任務名稱是 EventBridge 中的排程名稱。
+ 使用 Amazon EventBridge 和 AWS Step Functions 協調資料品質規則的評估和建議。