Optimice las canalizaciones de datos: cómo utilizar WhyLogs con PySpark para una validación y creación de perfiles de datos eficaces | de Sarthak Sarbahi | enero de 2024

Componentes de los registros de porqué

Comencemos por comprender las características importantes de los Whylogs.

• Registro de datos: El núcleo de Whylogs es su capacidad para registrar datos. Piense en ello como llevar un diario detallado de las características de sus datos. Registra varios aspectos de sus datos, como cuántas filas tiene, el rango de valores en cada columna y otros detalles estadísticos.
• Perfiles de Whylogs: Una vez que se registran los datos, Whylogs crea “perfiles”. Estos perfiles son como instantáneas que resumen sus datos. Incluyen estadísticas como promedios, recuentos y distribuciones. Esto es útil para comprender sus datos de un vistazo y realizar un seguimiento de cómo cambian con el tiempo.
• Seguimiento de datos: Con Whylogs, puede realizar un seguimiento de los cambios en sus datos a lo largo del tiempo. Esto es importante porque los datos a menudo evolucionan y lo que era cierto el mes pasado puede no serlo hoy. El seguimiento le ayuda a detectar estos cambios y comprender su impacto.
• Validación de datos: Whylogs le permite configurar reglas o restricciones para garantizar que sus datos sean los esperados. Por ejemplo, si sabe que una determinada columna solo debe tener números positivos, puede establecer una regla para ello. Si algo no coincide con sus reglas, sabrá que puede haber un problema.
• Visualización: Es más fácil comprender los datos a través de elementos visuales. Whylogs puede crear gráficos y tablas para ayudarle a ver lo que sucede en sus datos, haciéndolos más accesibles, especialmente para aquellos que no son expertos en datos.
• Integraciones: Whylogs admite integraciones con una variedad de herramientas, marcos y lenguajes: Spark, Kafka, Pandas, MLFlow, acciones de GitHub, RAPIDS, Java, Docker, AWS S3 y más.

Esto es todo lo que necesitamos saber sobre los Whylogs. Si tienes curiosidad por saber más, te animo a que consultes el documentación. A continuación, trabajemos para configurar todo para el tutorial.

Configuración del entorno

Usaremos un cuaderno Jupyter para este tutorial. Para que nuestro código funcione en cualquier lugar, usaremos JupyterLab en Docker. Esta configuración instala todas las bibliotecas necesarias y prepara los datos de muestra. Si eres nuevo en Docker y quieres aprender cómo configurarlo, consulta esto enlace.

Comience descargando los datos de muestra (CSV) de aquí. Estos datos son los que usaremos para la elaboración de perfiles y la validación. Crear un data carpeta en el directorio raíz de su proyecto y guarde el archivo CSV allí. A continuación, cree un Dockerfile en el mismo directorio raíz.

Este Dockerfile es un conjunto de instrucciones para crear un entorno específico para el tutorial. Vamos a desglosarlo:

• la primera linea FROM quay.io/jupyter/pyspark-notebook le dice a Docker que use una imagen existente como punto de partida. Esta imagen es un cuaderno Jupyter que ya tiene PySpark configurado.
• El RUN pip install whylogs whylogs[viz] whylogs[spark] La línea trata de agregar las bibliotecas necesarias a este entorno. Usa pip agregar whylogs y sus características adicionales para la visualización (viz) y para trabajar con Spark (spark).
• La última línea, COPY data/patient_data.csv /home/patient_data.csv, se trata de mover su archivo de datos a este entorno. Toma el archivo CSV patient_data.csv desde el data carpeta en el directorio de su proyecto y la coloca en el /home/ directorio dentro del entorno Docker.

A estas alturas el directorio de su proyecto debería verse así.

¡Impresionante! Ahora, creemos una imagen de Docker. Para hacer esto, escriba el siguiente comando en su terminal, asegurándose de estar en la carpeta raíz de su proyecto.

docker build -t pyspark-whylogs .

Este comando crea una imagen de Docker llamada pyspark-whylogs. Puedes verlo en la pestaña ‘Imágenes’ de tu Escritorio acoplable aplicación.

Siguiente paso: ejecutemos esta imagen para iniciar JupyterLab. Escribe otro comando en tu terminal.

docker run -p 8888:8888 pyspark-whylogs

Este comando lanza un contenedor desde el pyspark-whylogs imagen. Se asegura de que pueda acceder a JupyterLab a través del puerto 8888 de su computadora.

Después de ejecutar este comando, verá una URL en los registros similar a esta: http://127.0.0.1:8888/lab?token=your_token. Haga clic en él para abrir la interfaz web de JupyterLab.

¡Excelente! Todo está configurado para usar Whylogs. Ahora, conozcamos el conjunto de datos con el que trabajaremos.

Comprender el conjunto de datos

Usaremos un conjunto de datos sobre pacientes hospitalarios. El archivo, llamado patient_data.csvincluye 100k filas con estas columnas:

• patient_id: ID único de cada paciente. Recuerde, es posible que vea el mismo ID de paciente más de una vez en el conjunto de datos.
• patient_name: El nombre del paciente. Diferentes pacientes pueden tener el mismo nombre.
• height: La altura del paciente en centímetros. Cada paciente tiene la misma altura indicada para cada visita al hospital.
• weight: El peso del paciente en kilogramos. Siempre es más que cero.
• visit_date: La fecha en que el paciente visitó el hospital, en el formato YYYY-MM-DD.

En cuanto al origen de este conjunto de datos, no se preocupe. Fue creado por ChatGPT. A continuación, comencemos a escribir algo de código.

Empezando con PySpark

Primero, abra un nuevo cuaderno en JupyterLab. Recuerda guardarlo antes de empezar a trabajar.

Comenzaremos importando las bibliotecas necesarias.

# Import libraries
from typing import Any
import pyspark
from pyspark.sql import SparkSession
import pyspark.sql.functions as F
from whylogs.api.pyspark.experimental import collect_column_profile_views
from whylogs.api.pyspark.experimental import collect_dataset_profile_view
from whylogs.core.metrics.condition_count_metric import Condition
from whylogs.core.relations import Predicate
from whylogs.core.schema import DeclarativeSchema
from whylogs.core.resolvers import STANDARD_RESOLVER
from whylogs.core.specialized_resolvers import ConditionCountMetricSpec
from whylogs.core.constraints.factories import condition_meets
from whylogs.core.constraints import ConstraintsBuilder
from whylogs.core.constraints.factories import no_missing_values
from whylogs.core.constraints.factories import greater_than_number
from whylogs.viz import NotebookProfileVisualizer
import pandas as pd
import datetime

Luego, configuraremos una SparkSession. Esto nos permite ejecutar el código PySpark.

# Initialize a SparkSession
spark = SparkSession.builder.appName('whylogs').getOrCreate()
spark.conf.set("spark.sql.execution.arrow.pyspark.enabled","true")

Después de eso, crearemos un marco de datos de Spark leyendo el archivo CSV. También revisaremos su esquema.

# Create a dataframe from CSV file
df = spark.read.option("header",True).option("inferSchema",True).csv("/home/patient_data.csv")
df.printSchema()

A continuación, echemos un vistazo a los datos. Veremos la primera fila en el marco de datos.

# First row from dataframe
df.show(n=1, vertical=True)

Ahora que hemos visto los datos, es hora de comenzar a crear perfiles de datos con Whylogs.

Elaboración de perfiles de datos con Whylogs

Para perfilar nuestros datos, utilizaremos dos funciones. Primero, hay collect_column_profile_views. Esta función recopila perfiles detallados para cada columna del marco de datos. Estos perfiles nos brindan estadísticas como recuentos, distribuciones y más, dependiendo de cómo configuramos los registros de por qué.

# Profile the data with whylogs
df_profile = collect_column_profile_views(df)
print(df_profile)

Cada columna del conjunto de datos tiene su propia ColumnProfileView objeto en un diccionario. Podemos examinar varias métricas para cada columna, como sus valores medios.

Whylogs analizará cada punto de datos y decidirá estadísticamente si ese punto de datos es relevante o no para el cálculo final.

Por ejemplo, veamos el promedio height.

df_profile["height"].get_metric("distribution").mean.value

A continuación, también calcularemos la media directamente desde el marco de datos para comparar.

# Compare with mean from dataframe
df.select(F.mean(F.col("height"))).show()

Pero crear perfiles de columnas una por una no siempre es suficiente. Entonces, usamos otra función, collect_dataset_profile_view. Esta función perfila todo el conjunto de datos, no solo columnas individuales. Podemos combinarlo con Pandas para analizar todas las métricas del perfil.

# Putting everything together
df_profile_view = collect_dataset_profile_view(input_df=df)
df_profile_view.to_pandas().head()

También podemos guardar este perfil como un archivo CSV para su uso posterior.

# Persist profile as a file
df_profile_view.to_pandas().reset_index().to_csv("/home/jovyan/patint_profile.csv",header = True,index = False)

La carpeta /home/jovyan en nuestro contenedor Docker es de Pilas Docker de Jupyter (imágenes de Docker listas para usar que contienen aplicaciones Jupyter). En estas configuraciones de Docker, ‘jovyan’ es el usuario predeterminado para ejecutar Jupyter. El /home/jovyan La carpeta es donde generalmente comienzan los cuadernos de Jupyter y donde debe colocar los archivos para acceder a ellos en Jupyter.

Y así es como perfilamos los datos con Whylogs. A continuación, exploraremos la validación de datos.

Validación de datos con Whylogs

Para nuestra validación de datos, realizaremos estas comprobaciones:

• patient_id: Asegúrese de que no falten valores.
• weight: Asegúrese de que cada valor sea mayor que cero.
• visit_date: Compruebe si las fechas están en el YYYY-MM-DD formato.

Ahora comencemos. La validación de datos en Whylogs comienza con la elaboración de perfiles de datos. Podemos usar el collect_dataset_profile_view función para crear un perfil, como vimos antes.

Sin embargo, esta función suele crear un perfil con métricas estándar como promedio y recuento. Pero ¿y si necesitamos comprobar? valores individuales en una columna, a diferencia de las otras restricciones, ¿eso se puede comparar con métricas agregadas? Ahí es donde entran las métricas de recuento de condiciones. Es como agregar una métrica personalizada a nuestro perfil.

Creemos uno para el visit_date columna para validar cada fila.

def check_date_format(date_value: Any) -> bool:
date_format = '%Y-%m-%d'
try:
datetime.datetime.strptime(date_value, date_format)
return True
except ValueError:
return False
visit_date_condition = {"is_date_format": Condition(Predicate().is_(check_date_format))}

Una vez que tenemos nuestra condición, la agregamos al perfil. Usamos un Esquema estándar y agregue nuestro cheque personalizado.

# Create condition count metric
schema = DeclarativeSchema(STANDARD_RESOLVER)
schema.add_resolver_spec(column_name="visit_date", metrics=[ConditionCountMetricSpec(visit_date_condition)])

Luego volvemos a crear el perfil con métricas estándar y nuestra nueva métrica personalizada para el visit_date columna.

# Use the schema to pass to logger with collect_dataset_profile_view
# This creates profile with standard metrics as well as condition count metrics
df_profile_view_v2 = collect_dataset_profile_view(input_df=df, schema=schema)

Con nuestro perfil listo, ahora podemos configurar nuestras comprobaciones de validación para cada columna.

builder = ConstraintsBuilder(dataset_profile_view=df_profile_view_v2)
builder.add_constraint(no_missing_values(column_name="patient_id"))
builder.add_constraint(condition_meets(column_name="visit_date", condition_name="is_date_format"))
builder.add_constraint(greater_than_number(column_name="weight",number=0))
constraints = builder.build()
constraints.generate_constraints_report()

También podemos utilizar Whylogs para mostrar un informe de estas comprobaciones.

# Visualize constraints report using Notebook Profile Visualizer
visualization = NotebookProfileVisualizer()
visualization.constraints_report(constraints, cell_height=300)

Será un informe HTML que mostrará qué comprobaciones se aprobaron o no.

Esto es lo que encontramos:

• El patient_id A la columna no le faltan valores. ¡Bien!
• Alguno visit_date los valores no coinciden con YYYY-MM-DD formato.
• Algunos weight los valores son cero.

Verifiquemos dos veces estos hallazgos en nuestro marco de datos. Primero, comprobamos el visit_date formato con código PySpark.

# Validate visit_date column
df \
.withColumn("check_visit_date",F.to_date(F.col("visit_date"),"yyyy-MM-dd")) \
.withColumn("null_check",F.when(F.col("check_visit_date").isNull(),"null").otherwise("not_null")) \
.groupBy("null_check") \
.count() \
.show(truncate = False)
+----------+-----+
|null_check|count|
+----------+-----+
|not_null  |98977|
|null      |1023 |
+----------+-----+

Muestra que 1023 de 100.000 filas no coinciden con nuestro formato de fecha. A continuación, el weight columna.

# Validate weight column
df \
.select("weight") \
.groupBy("weight") \
.count() \
.orderBy(F.col("weight")) \
.limit(1) \
.show(truncate = False)
+------+-----+
|weight|count|
+------+-----+
|0     |2039 |
+------+-----+

Nuevamente, nuestros hallazgos coinciden con los registros de por qué. Casi 2000 filas tienen un peso de cero. Y eso concluye nuestro tutorial. Puedes encontrar el cuaderno para este tutorial. aquí.