Aplicar expectativas a nuevos datos

Introducción a la calidad de datos con Great Expectations

Davina Moossazadeh

Data Scientist

Checkpoints

Checkpoint: objeto que agrupa y ejecuta Definiciones de validación con parámetros compartidos

Un esquema que muestra un Checkpoint con este flujo: Solicitudes de lote -> Fuente de datos -> Definición de validación. La Definición produce Resultados de validación, que alimentan una Lista de acciones opcional con una o más Acciones.

Acciones: componentes configurados por Checkpoints que integran GX con otras herramientas según los Resultados de validación

¹ https://docs.greatexpectations.io/docs/core/trigger_actions_based_on_results/create_a_checkpoint_with_actions/

¿Por qué usar Checkpoints?

Reutilización

Permite ejecutar varias Definiciones de validación contra un Lote

Acciones

Permite activar Acciones según los Resultados de validación

Crear un Checkpoint

Crear un Checkpoint con notificación de Slack vía gx.Checkpoint():

checkpoint = gx.Checkpoint(

    name="my_checkpoint",

    validation_definitions=[validation_definition],

    actions=[SlackNotificationAction()]  # opcional
)

Errores de Checkpoint

Ejecutar un Checkpoint antes de añadir la Definición de validación al Contexto de datos genera un error:

CheckpointRelatedResourcesFreshnessError: 
ValidationDefinition 'my_validation_definition' must be added to the DataContext 
before it can be updated. Please call `context.validation_definitions.add(
<VALIDATION_DEFINITION_OBJECT>)`, then try your action again.

Añadir una Definición de validación

Añade la Definición de validación al Contexto de datos con .validation_definitions.add():

validation_definition = context.validation_definitions.add(

    validation_definition=validation_definition
)

Ejecutar un Checkpoint

checkpoint_results = checkpoint.run(
    batch_parameters={"dataframe": dataframe}
)

Resultados del Checkpoint. La salida es larga y no cabe en la diapositiva.

Evaluar resultados del Checkpoint

print(checkpoint_results.success)

False

print(checkpoint_results.describe())

Evaluar resultados del Checkpoint

{ "success": false,
  "statistics": {
    "evaluated_expectations": 1, "successful_expectations": 0,
    "unsuccessful_expectations": 1, "success_percent": 0.0
  },
  "expectations": [{
    "expectation_type": "expect_table_row_count_to_equal",
    "success": false,
    "kwargs": {"batch_id": ""my_datasource-my_dataframe_asset", "value": 118000}, 
    "result": {"observed_value": 11866}}
  ],
  "result_url": "https://app.greatexpectations.io/organizations/my_org/data-assets/*/validations/expectation-suites/0a123b9c-e370-4b18-b703-785dde88732d/results/cb093105-6ede-47d4-a141-dee10c632e18"
}

Data Docs

Data Docs: sitios estáticos generados a partir de metadatos de GX

# Checkpoint con acción para actualizar Data Docs
gx.Checkpoint(
    name,
    validation_definitions,
    actions=[
      gx.checkpoint.actions.UpdateDataDocsAction(
          name="update_my_site", site_names="my_data_docs_site"
        )
    ],
)

¹ https://docs.greatexpectations.io/docs/core/configure_project_settings/configure_data_docs/

Data Docs

Captura de Data Docs con una lista de Expectativas a nivel de tabla y sus valores observados. Una Expectativa de rango de filas muestra el Historial de validación expandido con hora de ejecución, valor observado y valores mín. y máx. Arriba a la derecha, un cuadro verde con una marca indica "Todas las Expectativas cumplidas".

Guía rápida

Añadir la Definición de validación al Contexto de datos:

context.validation_definitions.add(
    validation_definition
)

Crear Checkpoint:

checkpoint = gx.Checkpoint(
    name: str, 
    validation_definitions: list,
)

Ejecutar el Checkpoint:

checkpoint_results = checkpoint.run(
    batch_parameters={"dataframe": dataframe}
)

Comprobar resultados del Checkpoint:

checkpoint_results.success
checkpoint_results.describe()

¡Vamos a practicar!

Introducción a la calidad de datos con Great Expectations