Configuración de alertas y notificaciones en la EventBridge integración de Amazon Configure alertas y notificaciones en la CloudWatch integración Cómo consultar de resultados de calidad de datos Implementación de reglas de calidad de datos Programación de reglas de calidad de datos

Configuración de alertas, implementaciones y programación

En este tema se describe cómo configurar alertas, despliegues y programación para AWS Glue Data Quality.

Contenido

Configuración de alertas y notificaciones en la EventBridge integración de Amazon
- Opciones de configuración adicionales para el patrón de eventos
- Formatación de notificaciones como correos electrónicos
Configure alertas y notificaciones en la integración CloudWatch
Cómo consultar los resultados de calidad de los datos para crear paneles
Implementación de reglas de calidad de datos mediante AWS CloudFormation
Programación de reglas de calidad de datos

Configuración de alertas y notificaciones en la EventBridge integración de Amazon

AWS Glue Data Quality admite la publicación de EventBridge eventos, que se emiten al completar una evaluación del conjunto de reglas de calidad de datos. De este modo, puede configurar fácilmente alertas cuando fallen las reglas de calidad de datos.

Este es un ejemplo de un evento al evaluar los conjuntos de reglas de calidad de los datos en el Data Catalog. Con esta información, puedes revisar los datos que están disponibles en Amazon EventBridge. Puede realizar llamadas a la API adicionales para obtener más información. Por ejemplo, llama a la API get_data_quality_result con el ID de resultado para obtener los detalles de una ejecución concreta.


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

Este es un ejemplo de evento que se publica al evaluar los conjuntos de reglas de calidad de los datos en las libretas AWS Glue ETL o AWS Glue Studio.


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

Para que la evaluación de la calidad de los datos se ejecute tanto en el catálogo de datos como en los trabajos de ETL, la Amazon CloudWatch opción Publicar métricas en, que está seleccionada de forma predeterminada, debe permanecer seleccionada para que la EventBridge publicación funcione.

Configurar las EventBridge notificaciones

Propiedades de calidad de datos en AWS CloudFormation

Para recibir los eventos emitidos y definir los objetivos, debes configurar EventBridge las reglas de Amazon. Para crear reglas:

Abre la EventBridge consola de Amazon.
Elija Reglas en la sección Buses de la barra de navegación.
Elija Create Rule.
En Definir los detalles de las reglas:
1. En Nombre, escriba myDQRule.
2. Ingrese la descripción (opcional).
3. Para el bus de eventos, seleccione su bus de eventos. Si no tiene uno, déjelo como predeterminado.
4. En Tipo de regla, seleccione Regla con un patrón de evento y luego elija Siguiente.
En Construir patrón de evento:
1. Para la fuente del evento, selecciona AWS eventos o eventos EventBridge asociados.
2. Omita la sección de eventos de muestra.
3. Para el método de creación, seleccione Usar forma de patrón.
4. Para el patrón de eventos:
  1. Seleccione Servicios de AWS para el origen del evento.
  2. Seleccione Glue Data Quality para obtener AWS servicio.
  3. Seleccione Resultados de la evaluación de la calidad de los datos disponibles para el tipo de evento.
  4. Seleccione FALLA para estado(s) específico(s). A continuación, puede ver un patrón de eventos similar al siguiente:
    
    { "source": ["aws.glue-dataquality"], "detail-type": ["Data Quality Evaluation Results Available"], "detail": { "state": ["FAILED"] } }
  5. Para más opciones de configuración, consulte Opciones de configuración adicionales para el patrón de eventos.
En Objetivo(s) seleccionado(s):
1. Para Tipos de objetivos, seleccione el Servicio de AWS .
2. Utilice el menú desplegable Seleccione un destino para elegir el AWS servicio al que desee conectarse (SNS, Lambda, SQS, etc.) y, a continuación, seleccione Siguiente.
En Configurar etiqueta(s), haga clic en Agregar nuevas etiquetas para agregar etiquetas opcionales y, a continuación, seleccione Siguiente.
Aparecerá una página de resumen de todas las selecciones. Selecciona Crear regla en la parte inferior.

Opciones de configuración adicionales para el patrón de eventos

Además de filtrar el evento en función del éxito o el fracaso, puede que desee filtrar más los eventos según distintos parámetros.

Para ello, vaya a la sección Patrón de eventos y seleccione Editar patrón para especificar parámetros adicionales. Tenga en cuenta que los campos del patrón de eventos distinguen entre mayúsculas y minúsculas. A continuación, se muestran ejemplos de cómo configurar el patrón de eventos.

Para capturar eventos de una tabla en particular que evalúe conjuntos de reglas específicos, utilice este tipo de patrón:


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

Para capturar eventos de trabajos específicos de la experiencia ETL, utilice este tipo de patrón:


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

Para capturar eventos con una puntuación inferior a un umbral específico (por ejemplo, 70 %):


{
  "source": ["aws.glue-dataquality"],
  "detail-type": ["Data Quality Evaluation Results Available"],
  "detail": {
    "score": [{
      "numeric": ["<=", 0.7]
    }]
  }
}

Formatación de notificaciones como correos electrónicos

A veces es necesario enviar una notificación por correo electrónico con un formato adecuado a los equipos de la empresa. Para ello, puede utilizar Amazon EventBridge y AWS Lambda.

Notificación de calidad de datos con formato de correo electrónico

El siguiente código de ejemplo se puede utilizar para formatear las notificaciones de calidad de los datos a fin de generar correos electrónicos.


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')
    }

Configure alertas y notificaciones en la integración CloudWatch

Nuestro enfoque recomendado es configurar alertas de calidad de datos con Amazon EventBridge, ya que Amazon EventBridge requiere una configuración única para alertar a los clientes. Sin embargo, algunos clientes prefieren Amazon CloudWatch debido a su familiaridad. Para estos clientes, ofrecemos la integración con Amazon CloudWatch.

Cada evaluación de calidad de datos de AWS Glue emite un par de métricas denominadas glue.data.quality.rules.passed (que indican el número de reglas que se han aprobado) y glue.data.quality.rules.failed (que indican el número de reglas no aprobadas) por ejecución de calidad de datos. Puede usar esta métrica emitida para crear alarmas que alerten a los usuarios si una determinada cantidad de calidad de datos cae por debajo de un umbral. Para empezar a configurar una alarma que envíe un correo electrónico a través de una notificación de Amazon SNS, siga los pasos que se indican a continuación:

Para empezar a configurar una alarma que envíe un correo electrónico a través de una notificación de Amazon SNS, siga los pasos que se indican a continuación:

Abre la CloudWatch consola de Amazon.
Seleccione Todas las métricas en Métricas. Verá un espacio de nombres adicional en Espacios de nombres personalizados denominado Calidad de datos de Glue.

nota
Al iniciar una ejecución de AWS Glue Data Quality, asegúrate de que la CloudWatch casilla Publicar métricas en Amazon esté activada. De lo contrario, las métricas de esa ejecución concreta no se publicarán en Amazon CloudWatch.

En el espacio de nombres Glue Data Quality, puede ver las métricas que se emiten por tabla y por conjunto de reglas. Para este tema, utilizaremos la glue.data.quality.rules.failed regla y la alarma si este valor supera 1 (lo que indica que, si vemos un número de evaluaciones de reglas fallidas superior a 1, queremos que se nos notifique).
Para crear la alarma, seleccione Todas las alarmas en Alarmas.
Elija Create alarm (Crear alarma).
Elija Select Metric (Seleccionar métrica).
Seleccione la métrica glue.data.quality.rules.failed correspondiente a la tabla que creó y, a continuación, elija Seleccionar métrica.
En la pestaña Especificar métrica y condiciones, en la sección Métricas:
1. En Statistic (Estadística), elija Sum (Suma).
2. Para Periodo, seleccione 1 minuto.
En la sección Condiciones:
1. En Threshold type (Tipo de umbral), elija Static (Estático).
2. Para Siempre que glue.data.quality.rules.failed sea…, seleccione Mayor que/Igual a.
3. Para que..., ingrese 1 como valor umbral.
Estas selecciones implican que si la métrica glue.data.quality.rules.failed emite un valor mayor o igual a 1, activaremos una alarma. Sin embargo, si no hay datos, los consideraremos aceptables.
Elija Siguiente.
En Acciones de configuración:
1. Para la sección Desencadenador de estado de alarma, elija En alarma.
2. Para la sección Enviar una notificación al siguiente tema de SNS, seleccione Crear un tema nuevo para enviar una notificación a través de un nuevo tema de SNS.
3. En Puntos de conexión de correo electrónico que recibirán la notificación, ingrese una dirección de correo electrónico. Luego haga clic en Crear tema.
4. Elija Siguiente.
En Nombre de la alarma, ingrese myFirstDQAlarm y, a continuación, seleccione Siguiente.
Aparecerá una página de resumen de todas las selecciones. Seleccione Crear alarma en la parte inferior.

Ahora puedes ver la alarma que se está creando desde el panel de CloudWatch alarmas de Amazon.

Cómo consultar los resultados de calidad de los datos para crear paneles

Es posible que desee crear un panel para mostrar los resultados de calidad de sus datos. Hay dos formas de hacer esto:

Configure Amazon EventBridge con el siguiente código para escribir los datos en 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')
    }

Tras escribir en Amazon S3, puede utilizar los rastreadores de AWS Glue para registrarse en Athena y consultar las tablas.

Configure una ubicación de Amazon S3 durante una evaluación de la calidad de los datos:

Al ejecutar tareas de calidad de datos en el catálogo de datos de AWS AWS Glue o en la ETL de Glue, puede proporcionar una ubicación de Amazon S3 para escribir los resultados de calidad de los datos en Amazon S3. Puede usar la siguiente sintaxis para crear una tabla que haga referencia al objetivo para leer los resultados de calidad de los datos.

Tenga en cuenta que debe ejecutar las consultas CREATE EXTERNAL TABLE y MSCK REPAIR TABLE por separado.


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>;

Una vez que haya creado la tabla anterior, podrá ejecutar consultas analíticas con Amazon Athena.

Implementación de reglas de calidad de datos mediante AWS CloudFormation

Puede utilizarlas AWS CloudFormation para crear reglas de calidad de datos. Para obtener más información, consulte AWS CloudFormation AWS Glue.

Programación de reglas de calidad de datos

Puede programar las reglas de calidad de los datos mediante los métodos siguientes:

Programe las reglas de calidad de los datos desde el catálogo de datos: los usuarios de ningún código pueden utilizar esta opción para programar fácilmente sus escaneos de calidad de datos. AWS Glue Data Quality creará el cronograma en Amazon EventBridge. Para programar las reglas de calidad de los datos:
- Navegue hasta el conjunto de reglas y haga clic en Ejecutar.
- En la Frecuencia de ejecución, seleccione la programación deseada y proporcione un Nombre de tarea. El nombre de esta tarea es el nombre de su programa EventBridge.
Utilice Amazon EventBridge y AWS Step Functions para organizar las evaluaciones y recomendaciones de las normas de calidad de los datos.

Aviso JavaScript está desactivado o no está disponible en su navegador.

Para utilizar la documentación de AWS, debe estar habilitado JavaScript. Para obtener más información, consulte las páginas de ayuda de su navegador.

Convenciones del documento

Uso de las API para medir y gestionar la calidad de los datos

Cifrado de datos en reposo para Calidad de datos de AWS Glue