Capítulo 4. Trabajar con almacenes de datos y funciones

Este trabajo se ha traducido utilizando IA. Agradecemos tus opiniones y comentarios: translation-feedback@oreilly.com

El aprendizaje automático toma los datos y los convierte en lógica predictiva. Los datos son esenciales para el proceso, pueden proceder de muchas fuentes y deben procesarse para que sean utilizables. Por tanto, la gestión y el procesamiento de los datos son los componentes más críticos del aprendizaje automático. Los datos pueden proceder de distintas fuentes:

Archivos: Datos almacenados en archivos locales o en la nube
Almacenes de datos: Bases de datos que alojan transacciones de datos históricos
Bases de datos en línea: Bases de datos SQL, NoSQL, gráficas o de series temporales que alojen datos transaccionales o de aplicaciones actualizados
Flujos de datos: Almacenamiento intermedio que aloja eventos y mensajes en tiempo real (para pasar datos de forma fiable entre servicios).
Servicios en línea: Cualquier servicio en la nube que pueda proporcionar datos valiosos (esto puede incluir servicios sociales, financieros, gubernamentales y de noticias).
Mensajes entrantes: Mensajes y notificaciones asíncronos, que pueden llegar a través de correos electrónicos o cualquier otro servicio de mensajería (Slack, WhatsApp, Teams)

Los datos de origen se procesan y almacenan como características para su uso en el entrenamiento del modelo y en los flujos del modelo. En muchos casos, las características se almacenan en dos sistemas de almacenamiento: uno para el acceso por lotes (entrenamiento, predicción por lotes, etc.) y otro para la recuperación en línea (para el servicio en tiempo real o en línea). Como resultado, puede haber dos conductos de procesamiento de datos separados, uno que utilice el procesamiento por lotes y otro que utilice el procesamiento en tiempo real (flujo).

Es probable que las fuentes de datos y la lógica de procesamiento cambien con el tiempo, lo que provocará cambios en las características procesadas y en el modelo producido a partir de esos datos. Por tanto, aplicar el control de versiones a los datos, la lógica de procesamiento y el seguimiento del linaje de los datos son elementos críticos en cualquier solución MLOps.

La entrega de modelos de producción precisos y de alta calidad requiere grandes volúmenes de datos y una potencia de procesamiento significativa. El procesamiento de los datos de producción puede escalarse utilizando motores de análisis distribuidos(Apache Spark, Dask, Google BigQuery, etc.), tecnologías de procesamiento de flujos (como Apache Flink) o canalizaciones de datos de varias etapas.

Uno de los mecanismos para automatizar la integración con fuentes de datos, el procesamiento escalable de datos por lotes y en tiempo real, el versionado de datos y la gestión de funciones es utilizar un almacén de funciones. Un almacén de características es un eje central para producir, compartir y monitorear características. Los almacenes de características son esenciales en las implementaciones modernas de MLOps y se describirán con más detalle en este capítulo.

Versionado y linaje de datos

Los modelos y productos de datos se derivan de los datos. Por tanto, recopilar metadatos y rastrear el origen de los datos permite mejorar el control y la gobernanza de los productos de datos. Además, si quieres examinar una versión concreta de un producto de datos, debes conocer los datos originales utilizados para elaborar ese producto o modelo.

El versionado de datos, el linaje y la gestión de metadatos son un conjunto de prácticas MLOps esenciales que abordan lo siguiente:

Calidad de los datos: Rastrear los datos a través de los sistemas de una organización y recopilar metadatos e información sobre el linaje puede ayudar a identificar errores e incoherencias. Esto permite tomar medidas correctoras y mejorar la calidad de los datos.
Reproducibilidad y trazabilidad del modelo: El acceso a las versiones históricas de los datos nos permite reproducir los resultados del modelo y puede utilizarse para depurar el modelo, solucionar problemas y probar diferentes conjuntos de parámetros.
Gobernanza y auditabilidad de los datos: Al conocer el origen y el historial de los datos, las organizaciones pueden asegurarse de que los datos siguen las políticas y normativas previstas, rastrear las fuentes de errores y realizar auditorías o investigaciones.
Conformidad: El linaje de datos puede ayudar a las organizaciones a demostrar el cumplimiento de normativas como el GDPR y la HIPAA.
Gestión de datos más sencilla: Los metadatos y la información sobre el linaje permiten descubrir, mapear, perfilar, integrar y migrar mejor los datos.
Mejor colaboración: El versionado y el linaje de los datos pueden facilitar la cooperación entre los científicos de datos y los ingenieros de ML, al proporcionar una visión clara y coherente de los datos utilizados en los modelos de ML y al gestionar las actualizaciones.
Seguimiento de la dependencia: Comprender cómo cada cambio de datos, parámetros o código contribuye a los resultados y proporcionar información sobre qué datos u objetos del modelo deben cambiar debido a la modificación de la fuente de datos.

Cómo funciona

Como se muestra en la Figura 4-1, el flujo de generación de datos puede abstraerse como un conjunto de fuentes de datos y parámetros que se utilizan como entradas para una tarea de procesamiento de datos(computación) que produce una colección de productos de datos o artefactos. Los artefactos de salida pueden ser de distintos tipos, archivos, tablas, modelos de aprendizaje automático, gráficos, etc.

El sistema de seguimiento de datos registra la información sobre las entradas (fuentes y versiones de datos, parámetros) y las tareas de cálculo (código, paquetes, recursos, usuario ejecutor, etc.). Luego, la añade como metadatos en los artefactos de salida. Los metadatos pueden incluir información adicional como etiquetas o tags proporcionados por el usuario, información sobre la estructura de datos, esquema, estadísticas, etc. Normalmente, los metadatos no se copian en cada artefacto de salida, sino que se hace referencia a ellos (mediante un enlace) para eliminar la duplicación de datos.

Como se muestra en la Figura 4-2, los artefactos de salida de la primera tarea (por ejemplo, la preparación de datos) pueden utilizarse como datos de entrada para las siguientes tareas (por ejemplo, el entrenamiento del modelo, la prueba).

Al acceder a un producto de datos a través de una interfaz de usuario o un SDK, los metadatos nos permiten ver las fuentes de datos exactas, los parámetros y todos los detalles de la tarea de cálculo. También podemos seguir el progreso de los datos generados en un flujo multietapa y examinar todos los metadatos adicionales.

Cada vez que se ejecuta la tarea de procesamiento de datos, se crea una nueva versión de los artefactos de salida (ver Figura 4-3). Cada versión se marca con un identificador de versión único (commit id) y también puede etiquetarse con un nombre de versión significativo, como master, desarrollo, staging, producción, etc. Esto es similar al flujo de Git cuando se versiona código fuente.

Supongamos que estás ejecutando repetidamente una tarea específica cada hora. Tiene las mismas entradas y parámetros, o puedes hacer pequeños cambios que no modifican los resultados de los datos de salida. Esto puede dar lugar a grandes pilas de datos redundantes, y varias versiones almacenarán el mismo contenido. Muchas soluciones de versionado de datos implementan la deduplicación de contenidos para abordar este reto.

Cuando se produce un artefacto, se calcula un valor hash criptográfico del contenido (por ejemplo, utilizando los algoritmos MD5 o SHA1), que representa la unicidad del contenido. Finalmente, el valor hash se compara con versiones anteriores o se utiliza como índice en el sistema de almacenamiento. De esta forma, el contenido se almacena sólo una vez.

Dado que la naturaleza de las soluciones de versionado de datos es hacer un seguimiento de varios atributos además de los datos de origen (código, parámetros, usuarios, recursos, etc.), debe estar bien integrada con el sistema de control de origen (Git) y el marco de ejecución de trabajos o canalizaciones. De lo contrario, el usuario debe pegar manualmente los marcos y proporcionar los metadatos de referencia para registrarlos junto con los datos.

Muchos marcos de trabajo(MLflow, MLRun y otros) proporcionan una API de registro, en la que el usuario llama a un método log_artifact(), que automáticamente registra y versiona los nuevos datos junto con el código y los metadatos de ejecución. Muchos podrían ofrecer una solución auto logging que no requiere instrumentación del código. En su lugar, averiguará automáticamente qué datos y metadatos necesitan ser guardados y versionados mediante la comprensión del código del usuario y de las capacidades del marco ML.

Herramientas comunes de versionado de datos de ML

Existe un conjunto de marcos comerciales y de código abierto para el versionado de datos. Este libro se centra en explicar y comparar las opciones de código abierto DVC, Pachyderm, MLflow y MLRun.

Control de versiones de datos

El Control de Versiones de Datos (DVC) comenzó como una herramienta de versionado de datos para ML y se amplió para dar soporte a la automatización básica del flujo de trabajo de ML y a la gestión de experimentos. Aprovecha el conjunto de herramientas de ingeniería de software con las que ya estás familiarizado (Git, tu IDE, CI/CD, etc.).

DVC funciona igual que Git (con comandos similares), pero para grandes conjuntos de datos basados en archivos y artefactos de modelos. Ésta es su principal ventaja, pero también su debilidad. DVC almacenael contenido de los datos en archivos o en un almacenamiento de objetos (AWS S3, GCS, Azure Blob, etc.) y guarda una referencia a esos objetos en un archivo (.dvc), que se almacena en el repositorio Git.

El siguiente comando añadirá un archivo de modelo local(model.pkl) al sistema de versiones de datos:

dvc add model.pkl

DVC copiará el contenido del archivo model.pkl en un nuevo archivo con un nuevo nombre (derivado del valor hash md5 del contenido) y lo colocará bajo el directorio .dvc/. También crea un archivo llamado model.pkl.dvc, que apunta a ese archivo de contenido. A continuación, el nuevo archivo de metadatos debe ser rastreado por Git, el contenido debe ser ignorado y los cambios deben ser confirmados. Esto se hace escribiendo los siguientes comandos:

git add model.pkl.dvc .gitignore
git commit -m "Add raw data"

Cuando quieras subir los datos a tu almacenamiento remoto, tendrás que configurar un repositorio de objetos remoto (que no se muestra aquí) y utilizar el comando push de DVC:

dvc push

El flujo de datos se ilustra en la Figura 4-4.

Como puedes ver en el ejemplo, DVC proporciona una sincronización fiable entre el código y los objetos de datos de archivo, pero requiere una configuración manual y no almacena metadatos ampliados sobre la ejecución, el flujo de trabajo, los parámetros, etc. En su lugar, DVC gestiona los parámetros y las métricas de resultados mediante archivos JSON o YAML almacenados y versionados junto al código.

Los usuarios pueden definir etapas de flujo de trabajo que envuelvan un ejecutable (por ejemplo, un programa Python) y especificar qué parámetros (-p) son las entradas o dependencias de archivos (-d) y salidas (-o) a ese ejecutable (ver Ejemplo 4-1).

Ejemplo 4-1. Añadir un paso de flujo de trabajo en DVC

dvc stage add -n featurize \
              -p featurize.max_features,featurize.ngrams \
              -d src/featurization.py -d data/prepared \
              -o data/features \
              python src/featurization.py data/prepared data/features

Cuando ejecutes el comando dvc repro, evaluará si las dependencias han cambiado, ejecutará los pasos necesarios y registrará las salidas.

DVC no utiliza una base de datos de experimentos. Utiliza Git como base de datos, y cada ejecución o combinación de parámetros se asigna a un único commit de Git. Además, DVC se centra en el desarrollo local. Por lo tanto, utilizarlo a escala o en un entorno de flujo de trabajo en contenedores o distribuido puede ser un reto y requerir scripts e integraciones manuales.

En resumen, DVC es una herramienta excelente para versionar grandes artefactos de datos y asignarlos a commits de Git en un entorno de desarrollo local. Además, DVC implementa la deduplicación de datos para reducir la huella de almacenamiento real. Por otro lado, DVC está orientado a la línea de comandos (flujo Git) y tiene capacidades limitadas para funcionar en producción, ejecutar canalizaciones y realizar el seguimiento de atributos extendidos y datos estructurados. También viene con una interfaz de usuario mínima (estudio).

Paquidermo

Pachyderm es una herramienta de canalización y versionado de datos basada en una infraestructura de contenedores. Proporciona un sistema de archivos versionados y permite a los usuarios construir canalizaciones de varias etapas, en las que cada etapa se ejecuta en un contenedor, acepta datos de entrada (como archivos) y genera archivos de datos de salida.

Pachyderm proporciona un repositorio de datos versionados que puede implementarse sobre almacenamiento de objetos (por ejemplo, AWS S3, Minio, GCS) y al que se accede mediante una API de archivos o el SDK/CLI. Cada confirmación o cambio de datos se registra de forma similar a Git. Los datos se deduplican para preservar el espacio.

La canalización de datos Pachyderm ejecuta contenedores y monta una porción del repositorio en el contenedor (bajo el directorio /pfs/ ). El contenedor lee los archivos, los procesa y escribe los resultados en el repositorio Pachyderm.

El ejemplo 4-2 muestra una sencilla definición de canalización que toma todos los datos del repositorio data en la rama master, ejecuta la lógica de recuento de palabras (utilizando la imagen contenedora especificada) y escribe la salida en el repositorio out.

Ejemplo 4-2. Ejemplo de tubería Pachyderm

pipeline:
    name: 'count'
description: 'Count the number of lines in a csv file'
input:
    pfs:
        repo: 'data'
        branch: 'master'
        glob: '/'
transform:
    image: alpine:3.14.0
    cmd: ['/bin/sh']
    stdin: ['wc -l /pfs/data/iris.csv > /pfs/out/line_count.txt']

Las tuberías pueden activarse cada vez que cambien los datos de entrada, y los datos pueden procesarse de forma incremental (sólo se pasarán al proceso contenedor los archivos nuevos). Esto puede ahorrar tiempo y recursos.

Pachyderm tiene una bonita interfaz de usuario para gestionar las tuberías y explorar los datos. Véase la Figura 4-5.

Pachyderm puede trabajar con fuentes de datos estructuradas grandes o continuas, dividiendo los datos en archivos CSV o JSON más pequeños.

En resumen, Pachyderm es una herramienta excelente para construir pipelines de datos versionados, en los que el código es lo suficientemente sencillo como para leer y escribir archivos. Maneja la deduplicación y el procesamiento incremental. Sin embargo, requiere un seguimiento separado del código fuente (ejecuta imágenes preconstruidas), los parámetros de ejecución o experimento, los metadatos, las métricas, etc.

Seguimiento MLflow

MLflow es una plataforma de código abierto para gestionar el ciclo de vida integral del aprendizaje automático. Uno de sus componentes principales es MLflow Tracking, que proporciona una API y una interfaz de usuario para registrar las ejecuciones de aprendizaje automático, sus entradas y salidas, y visualizar los resultados. Las ejecuciones de MLflow Tracking son ejecuciones de algún código de ciencia de datos. Cada ejecución registra la siguiente información

Versión del código: Hash del commit de Git utilizado para la ejecución.
Hora de inicio y fin: La hora de inicio y fin de la carrera.
Fuente: El nombre del archivo para lanzar la ejecución, o el nombre del proyecto y el punto de entrada para la ejecución si se ejecuta desde un proyecto MLflow.
Parámetros: Key-value parámetros de entrada de tu elección. Tanto las claves como los valores son cadenas.
Métricas: Key-value métrica, cuyo valor es numérico. MLflow registra y te permite visualizar el historial completo de la métrica.
Artefactos: Da salida a archivos en cualquier formato. Por ejemplo, puedes grabar como artefactos imágenes (por ejemplo, PNG), modelos (por ejemplo, un modelo scikit-learn en escabeche) y archivos de datos (por ejemplo, un archivo Parquet).

El Seguimiento MLflow no es una solución completa de versionado de datos, ya que no admite funciones como el linaje de datos (registrar las entradas de datos y qué datos se utilizaron para crear un nuevo elemento de datos) o la deduplicación. Sin embargo, permite registrar e indexar las salidas de datos de cada ejecución junto con el código fuente, los parámetros y algunos detalles de la ejecución. MLflow puede integrarse manualmente con otras herramientas, como DVC, para realizar un seguimiento de los datos y los experimentos.

La ventaja de MLflow es el seguimiento de las salidas de datos con metadatos adicionales sobre el código y los parámetros, y su visualización o comparación en una interfaz gráfica de usuario. Sin embargo, esto no es gratuito. Es necesario instrumentar el código del usuario con el código de seguimiento de MLflow.

El Ejemplo 4-3 muestra un fragmento parcial de código que rastrea una ejecución utilizando la API MLflow. En primer lugar, los argumentos de la línea de comandos se analizan manualmente y los datos de entrada se pasan como una cadena URL, como cualquier otro parámetro. A continuación, la carga y la transformación de los datos se realizan manualmente.

Una vez ejecutada la lógica (preparación de datos, entrenamiento, etc.), el usuario registra las etiquetas, los parámetros de entrada, las métricas de salida y los artefactos de datos (conjunto de datos y modelo) mediante los comandos de registro de MLflow.

Ejemplo 4-3. Ejemplo de código de Seguimiento MLflow

if __name__ == "__main__":
    # parse the input parameters
    parser = argparse.ArgumentParser()
    parser.add_argument("--data", help="input data path", type=str)
    parser.add_argument('--dropout',  type=float, default=0.0, help='dropout ratio')
    parser.add_argument("--lr", type=float, default=0.001, help='learning rate')
    args = parser.parse_args()

    # Read the csv file
    try:
        data = pd.read_csv(args.data)
    except Exception as e:
        raise ValueError(f"Unable to read the training CSV, {e}")

    # additional initialization code ...

    with mlflow.start_run():

        # training code ...

        # log experiment tags, parameters and result metrics
        mlflow.set_tag("framework", "sklearn")
        mlflow.log_param("dropout", args.dropout)
        mlflow.log_param("lr", args.lr)
        mlflow.log_metric("rmse", rmse)
        mlflow.log_metric("r2", r2)
        mlflow.log_metric("mae", mae)

        # log data and model artifacts
        mlflow.log_artifacts(out_data_path, "output_data")
        mlflow.sklearn.log_model(model, "model",
                                 registered_model_name="ElasticnetWineModel")

MLflow envía la información de la ejecución al servidor de seguimiento y almacena los elementos de datos en archivos locales u objetos remotos (por ejemplo, en S3). La información de la ejecución puede visualizarse o compararse en la interfaz de usuario de MLflow (ver Figura 4-6).

MLflow no gestiona ni versiona objetos de datos. La ejecución es el elemento principal, y no puedes acceder ni buscar directamente objetos de datos y artefactos. Además, no hay seguimiento de linaje, lo que significa que no hay seguimiento de qué objetos de datos se utilizaron para producir un nuevo objeto de datos o artefacto. Cuando ejecutas una canalización, no puedes ver los artefactos de las distintas etapas en un mismo lugar ni encadenar la salida de una etapa a la entrada de la siguiente.

Con MLflow, la capacidad de almacenamiento puede llegar a ser significativa, ya que cada ejecución guarda los resultados en un nuevo directorio de archivos, aunque no haya cambiado nada. No hay deduplicación de datos como en los otros marcos.

En resumen, el seguimiento MLflow es una herramienta excelente para seguir y comparar los resultados de los experimentos de ML en un entorno de desarrollo. Además, MLflow es fácil de instalar y utilizar. Sin embargo, no es un sistema de seguimiento de datos ni de versiones, y puede requerir una capacidad de almacenamiento considerable. Además, MLflow requiere que los desarrolladores añadan código personalizado y que los equipos de MLOps añadan lógica de encolado para adaptarse a las Implementaciones de producción y a los flujos de trabajo CI/CD.

MLRun

MLRun es un marco de orquestación de MLOps de código abierto con múltiples subcomponentes para gestionar el ciclo de vida completo de ML. Los objetos de datos son ciudadanos de primera clase en MLRun y están bien integrados con los demás componentes para proporcionar una experiencia y automatización sin fisuras.

Mientras que la mayoría de los marcos gestionan objetos de datos de archivos, MLRun admite diversos objetos de datos (almacenes de datos, elementos/archivos, conjuntos de datos, flujos, modelos, conjuntos de características, vectores de características, gráficos, etc.), cada uno con metadatos, acciones y visores únicos.

Cada objeto de MLRun tiene un tipo, un ID de versión único, etiquetas (versiones con nombre como desarrollo, producción, etc.), etiquetas definidas por el usuario (para agrupar y buscar entre objetos) y relaciones con otros objetos, y es un miembro del proyecto. Por ejemplo, un objeto de ejecución tiene enlaces con los objetos de datos de origen y de salida y con los objetos de función (código), formando un gráfico de relaciones.

La Figura 4-7 muestra la pantalla de ejecución con pestañas de información para atributos generales y de código, objetos de entrada de datos, objetos de salida de datos/artefactos, métricas de resultados, registros auto-recopilados, etc. Los usuarios pueden ver la información desde distintas perspectivas. Por ejemplo, ver todos los conjuntos de datos del proyecto (independientemente de la ejecución que los haya generado).

Los objetos de datos y artefactos de MLRun llevan metadatos detallados, incluida información sobre cómo fueron producidos (por quién, cuándo, código, marco, etc.), qué fuentes de datos se utilizaron para crearlos y atributos específicos del tipo, como esquema, estadísticas, vista previa, etc. Los metadatos se autogeneran, lo que proporciona una mejor observabilidad y elimina la necesidad de lógica de encolado adicional.

Nota

Los usuarios de MLFlow pueden seguir utilizando MLFlow para el seguimiento de las API, y MLRun registrará automáticamente los datos, metadatos y modelos registrados como artefactos de producción, junto con metadatos y contexto operativos adicionales.

MLRun proporciona una amplia API/SDK para el seguimiento y la búsqueda en datos y experimentos. Sin embargo, el verdadero poder es que puede ofrecer la mayoría de las funciones y la automatización sin necesidad de codificación adicional.

El ejemplo 4-4 acepta datos y parámetros de entrada y genera datos y resultados de salida. Observa que, a diferencia de los ejemplos anteriores, el código no incluye el análisis sintáctico de argumentos, la carga de datos, la conversión, el registro, etc.

Ejemplo 4-4. Ejemplo de código MLRun

def data_preparation(dataset: pd.DataFrame, test_size=0.2):
    # preform processing on the dataset
    dataset = clean_df(dataset).dropna(how="any", axis="rows")
    dataset = dataset.drop(columns=["key", "pickup_datetime"])
    train, test = train_test_split(dataset, test_size=test_size)
    return train, test, "fare_amount"

Al ejecutar la función y especificar la URL o ruta del objeto de datos de entrada (un archivo, un objeto remoto o un tipo complejo), éste se carga automáticamente en la función. Por ejemplo, utilizando controladores AWS boto para acceder a objetos S3 o controladores BigQuery para acceder a una tabla BigQuery. A continuación, los datos se convierten al formato de aceptación (DataFrame) y se inyectan en el código de usuario.

MLRun puede autodetectar el tipo de valor devuelto (por ejemplo, train y test son de tipo DataFrame) y almacenarlo de la mejor forma, junto con metadatos autogenerados, enlaces a los detalles del trabajo y a los objetos de entrada de datos, e información sobre versiones. Si los datos se repiten, se deduplican y se almacenan sólo una vez para preservar el espacio de almacenamiento.

Los objetos de datos tienen un tipo específico de visualización en la UI y en el SDK cliente, independientemente de cómo y dónde estén almacenados; por ejemplo, formatos tabulares con metadatos de tabla (esquema, estadísticas y más) para los conjuntos de datos o gráficos interactivos para los objetos de gráfico (ver Figuras 4-8 y 4-9).

En resumen, MLRun es un completo marco de orquestación de MLOps, muy centrado en la gestión de datos, el movimiento, el versionado y la automatización. Además, MLRun tiene un rico modelo de objetos que abarca distintos tipos de datos y objetos de ejecución (funciones, ejecuciones, flujos de trabajo, etc.), cómo se relacionan y cómo se utilizan. MLRun se centra en la abstracción y la automatización para reducir los esfuerzos de desarrollo e implementación. Sin embargo, MLRun no es una solución general de gestión de datos y versiones, y su valor se maximiza cuando se utiliza en el contexto de MLOps.

Otros marcos

Algunas herramientas, como Delta Lake y lakeFS, gestionan el versionado del lago de datos. Sin embargo, esas herramientas no se centran en el ciclo de vida del ML y pueden requerir integración para que sean útiles para los MLOps.

Los proveedores en la nube ofrecen soluciones que suelen estar estrechamente vinculadas a sus servicios internos. Por ejemplo, consulta Rastreo de linajes ML de Amazon SageMaker y conjuntos de datos ML de Azure .

Preparación y análisis de datos a escala

El procesamiento de datos se utiliza ampliamente en los conductos de datos, ML y aplicaciones. Cuando se trabaja con datos de producción, existe la necesidad de soportar una escala y un rendimiento más amplios y, en algunos casos, manejar los datos a medida que llegan en tiempo real.

Las prácticas que funcionan durante el desarrollo interactivo, por ejemplo, almacenar los datos en un archivo CSV y leerlos en el cuaderno, no funcionan con gigabytes o terabytes de datos. Requieren enfoques de procesamiento de datos distribuidos o en paralelo.

La arquitectura general del procesamiento distribuido de datos es la misma, con diferencias en cómo se distribuyen y recopilan los datos y qué API utilizan. Por ejemplo, los datos se dividen en varios nodos informáticos, las solicitudes de procesamiento o consultas llegan a uno o más nodos para su procesamiento local, y los resultados se recogen y fusionan para obtener una única respuesta. Además, las consultas complejas pueden barajar datos entre nodos o ejecutar múltiples pasos de procesamiento y movimiento.

La Figura 4-10 muestra cómo funciona el procesamiento distribuido de datos utilizando el enfoque map-reduce para contar las palabras de un documento.

Transformaciones de datos estructurados y no estructurados

Los datos pueden ser estructurados, lo que significa que se ajustan a un formato o estructura específicos y a menudo tienen un esquema o modelo de datos predefinido. Los datos estructurados pueden ser una tabla de base de datos o archivos con un esquema estructurado (por ejemplo, CSV, Excel, JSON, ML, Parquet). Sin embargo, la mayoría de los datos del mundo son no estructurados, normalmente más complejos y más difíciles de procesar que los datos estructurados. Esto incluye texto libre, registros, páginas web, imágenes, vídeo y audio.

He aquí algunos ejemplos de transformaciones analíticas que pueden realizarse sobre datos estructurados:

Filtrado: Seleccionar un subconjunto de los datos en función de determinados criterios, como un intervalo de fechas concreto o valores específicos de una columna.
Clasificación: Ordenar los datos en función de una o varias columnas, como ordenar por fecha o por un valor concreto.
Agrupación: Organizar los datos en grupos basados en una o varias columnas, como agrupar por categoría de producto o por ciudad.
Agregación: Calcular estadísticas de resumen, como recuento, suma, media, máximo y desviación típica, para una o varias columnas.
Únete a: Combinar datos de varias tablas o conjuntos de datos basándose en columnas comunes, como unir una tabla de datos de ventas con una tabla de datos de clientes.
Cartografía: Mapear valores de una o más columnas a nuevos valores de columna mediante operaciones o código definidos por el usuario. El mapeo con estados puede calcular los nuevos valores basándose en los valores originales y en los estados acumulados de entradas anteriores (por ejemplo, el tiempo transcurrido desde el último inicio de sesión).
Análisis de series temporales: Analizar o agregar datos a lo largo del tiempo, como identificar tendencias, pautas o anomalías.

Las siguientes técnicas pueden utilizarse para procesar datos no estructurados o convertirlos en datos estructurados:

Minería de textos: Utilizar técnicas de PNL para extraer significado y conocimientos de los datos de texto. La minería de textos puede extraer información como sentimientos, entidades y temas de los datos de texto.
Visión por ordenador: Utiliza técnicas de procesamiento de imágenes y vídeos para extraer información de los datos visuales. La visión por ordenador puede extraer información como el reconocimiento de objetos, el reconocimiento facial y la clasificación de imágenes.
Reconocimiento de audio y voz: Utiliza técnicas de procesamiento de audio y voz a texto para extraer significado e información de los datos de audio. El reconocimiento de audio y voz puede extraer información como la conversión de voz a texto, el sentimiento y la identificación del hablante.
Extracción de datos: Utilizando técnicas como el web scraping y la extracción de datos para extraer datos estructurados de fuentes de datos no estructurados.

Se pueden utilizar varios métodos de ML para transformar los datos brutos en datos más significativos, por ejemplo:

Agrupación: Agrupar puntos de datos similares en función de determinadas características, como clientes con hábitos de compra similares
Reducción de la dimensionalidad: Reducir el número de características de un conjunto de datos para facilitar su análisis o visualización
Regresión y clasificación: Predecir una clase o un valor basándose en otras características de los datos
Imputar: Determinar el valor esperado basándose en otros puntos de datos en caso de que falten datos
Incrustar: Representar una secuencia de texto, audio o una imagen como un vector numérico que preserve las relaciones semánticas o las características contextuales.

Arquitecturas Distribuidas de Proceso de Datos

Las arquitecturas de procesamiento de datos pueden dividirse en tres categorías principales:

Interactivo: Llega una solicitud o una actualización, se procesa y se devuelve una respuesta inmediatamente; por ejemplo, bases de datos SQL y NoSQL, almacenes de datos, almacenes de claves/valores, bases de datos de gráficos, bases de datos de series temporales y servicios en la nube.
Lote: Un trabajo se inicia a partir de una solicitud o de una hora programada, los datos se obtienen y se procesan, y los resultados se escriben en el almacenamiento de destino una vez finalizados. Los trabajos por lotes suelen tardar más en procesarse. Algunos ejemplos de marcos para el procesamiento de datos por lotes son Hadoop, Spark y Dask.
Streaming: Procesamiento continuo de solicitudes entrantes o trozos de datos y escritura de los resultados en tiempo real en un almacenamiento de destino o cola de mensajes.

El procesamiento por lotes suele ser más eficaz para procesar grandes cantidades de datos. Sin embargo, el procesamiento interactivo y de flujo de datos proporciona respuestas más rápidas con retrasos más cortos. Además, construir canalizaciones de procesamiento de flujos de datos suele ser más complejo que los trabajos por lotes.

Algunos frameworks como Spark pueden admitir diferentes métodos de procesamiento (interactivo, por lotes, streaming), pero normalmente sólo serán más óptimos en uno de los métodos de procesamiento.

Tratamiento interactivo de datos

Se espera que los sistemas interactivos respondan inmediatamente, para que el cliente solicitante o el panel interactivo no tengan que esperar. Además, los servicios de producción pueden depender de la fiabilidad y solidez de los resultados. Por eso, los sistemas interactivos tienen API sencillas con operaciones de datos limitadas. En algunos casos, los sistemas interactivos proporcionan mecanismos para definir una lógica personalizada mediante procedimientos almacenados y funciones definidas por el usuario (UDF).

La principal diferencia entre los tipos de sistemas de datos interactivos es cómo indexan y almacenan los datos para minimizar el tiempo de recuperación de la respuesta. Por ejemplo, los almacenes NoSQL, en memoria y de clave/valor están optimizados para la recuperación mediante una clave de índice (como un id de usuario, id de producto, etc.). Los datos se dividen por la clave (o un hash criptográfico de la clave) y se almacenan en diferentes nodos. Cuando llega una solicitud, se pasa al nodo específico, que gestiona los datos de esa clave (usuario, producto, etc.) y puede calcular y recuperar rápidamente la respuesta. En cambio, los cálculos complejos o de clave cruzada requieren la coordinación entre todos los nodos y tardan mucho más.

Las bases de datos analíticas y los almacenes de datos están diseñados para recorrer muchos registros con distintos valores de clave de índice. Organizan los datos en columnas (por campos) y utilizan diversas tecnologías de compresión columnar y trucos de filtrado y sugerencias (como el filtrado bloom) para omitir bloques de datos.

Otros sistemas, como las bases de datos de series temporales o de gráficos, tienen disposiciones de datos y estrategias de búsqueda más avanzadas que combinan índices multidimensionales y compresión columnar. Por ejemplo, acceder al objeto métrica de series temporales por la clave métrica (nombre) y utilizar tecnologías de compresión columnar para explorar o agregar los valores individuales (por tiempo).

Muchos sistemas interactivos utilizan el lenguaje SQL o una semántica similar a SQL para procesar los datos.

En la Tabla 4-1 se enumeran algunas subcategorías de sistemas de datos notables.

Tabla 4-1. Categorías y descripciones de los sistemas de datos
Categoría	Descripción
Relacional	Almacena datos estructurados, accede a ellos mediante comandos SQL. Algunos ejemplos son MySQL, PostgreSQL, Oracle y Microsoft SQL Server.
NoSQL	Algunos ejemplos son MongoDB, Cassandra, Redis, Elasticsearch, AWS DynamoDB, Google BigTable y bases de datos no tabulares.
Series temporales	Almacena y consulta datos de series temporales. Algunos ejemplos son InfluxDB, Prometheus y TimescaleDB.
Gráfico	Almacena y consulta datos en formato gráfico. Algunos ejemplos son Neo4j y Titan.
Vector	Una base de datos vectorial indexa y almacena incrustaciones vectoriales de alta dimensión para una rápida recuperación y búsqueda de similitudes. Algunos ejemplos son Chroma, Pinecone, Milvus, Weaviate y Pgvector.

Los sistemas analíticos suelen recorrer y procesar conjuntos de datos más grandes. En consecuencia, admiten transformaciones más amplias (filtrado, agrupación, unión, agregación, asignación, etc.) y funciones definidas por el usuario. Además, algunas pueden procesar y agregar datos de otras bases de datos o datos almacenados en archivos. Por ejemplo, soluciones como Spark SQL o PrestoDB tienen conectores a muchas fuentes de datos y pueden procesar consultas que abarcan muchos conjuntos de datos y están almacenados en diferentes sistemas.

Uno de los motores analíticos distribuidos basados en SQL más populares es PrestoDB y su proyecto continuador, Trino. Presto fue desarrollado inicialmente por Facebook y contribuido al código abierto. Más tarde, se bifurcó en proyectos como Trino y en productos comerciales como el servicio en la nube Amazon Athena. Trino tiene una larga lista de conectores de datos.

La Figura 4-11 ilustra las arquitecturas de Presto y Trino. Las consultas llegan a través de solicitudes HTTP, se analizan y el planificador y el programador las dividen en tareas más pequeñas que los trabajadores procesan y combinan.

Procesamiento de datos por lotes

El procesamiento de datos por lotes se utiliza cuando es necesario procesar grandes cantidades de datos y ejecutar una secuencia de transformaciones de datos, y el tiempo de procesamiento es una preocupación menor. En el procesamiento por lotes, los datos se leen y se dividen en trozos que se pasan a varios trabajadores para que los procesen. Una vez que el resultado está listo, se escribe en el sistema de destino. El procesamiento por lotes se utiliza a menudo para procesar grandes cantidades de datos históricos y generar el conjunto de datos para el entrenamiento de modelos ML.

Uno de los marcos de procesamiento de datos por lotes más conocidos fue Apache Hadoop, un marco de software de código abierto para el almacenamiento distribuido y el procesamiento a gran escala de tareas intensivas en datos. Hadoop fue desarrollado inicialmente por ingenieros de Yahoo! y se basaba en el modelo de programación MapReduce, que consta de dos funciones principales: Map y Reduce. La función Map toma un conjunto de datos de entrada y lo procesa en un conjunto de pares clave-valor intermedios, que luego se agrupan por clave y son procesados por la función Reduce para producir el resultado final.

Desde entonces, Hadoop ha sido sustituido por arquitecturas más modernas y nativas de la nube basadas en almacenamiento de objetos en la nube, infraestructura en contenedores y marcos de computación como Spark, Flink, Beam, Dask y otros.

Un uso cotidiano del procesamiento por lotes se encuentra en las tareas ETL. ETL se refiere a extraer datos de múltiples fuentes, transformarlos y cargarlos en una base de datos, almacén de datos o lago de datos de destino. ETL es un paso crucial en el proceso de integración de datos, ya que permite a las organizaciones extraer, limpiar y transformar datos de múltiples fuentes en un repositorio único y centralizado.

Las canalizaciones de procesamiento por lotes pueden ser complejas y tener múltiples pasos y dependencias. Apache Airflow es uno de los marcos de trabajo de código abierto más populares de para crear, programar y monitorizar canalizaciones de datos por lotes.

Airflow fue desarrollado inicialmente por Airbnb y ahora lo mantiene la Fundación del Software Apache. Proporciona una interfaz sencilla y fácil de usar para definir flujos de trabajo como DAGs de tareas, donde cada tarea representa un paso de procesamiento individual. Las tareas pueden escribirse en Python y ejecutarse en varios entornos, como localmente, sobre Kubernetes o en la nube.

Airflow también proporciona una interfaz de usuario basada en web (ver Figura 4-12) para gestionar y monitorear los flujos de trabajo, incluida la posibilidad de ver el estado de cada tarea, reintentar tareas fallidas y activar o programar tareas manualmente. También incluye funciones para gestionar y organizar flujos de trabajo, como definir dependencias entre tareas y configurar la lógica de reintento de tareas.

El Ejemplo 4-5 es un ejemplo de código Python que puede utilizarse para crear un DAG en Apache Airflow que lea datos de un archivo CSV, los procese y los escriba en un destino.

Ejemplo 4-5. Ejemplo de código de canalización de datos de flujo de aire

import csv
from airflow import DAG
from airflow.operators.python_operator import PythonOperator
from datetime import datetime, timedelta

def process_data(**kwargs):
    ti = kwargs['ti']
    input_file = ti.xcom_pull(task_ids='read_file')
    processed_data = do_data_processing(input_file)
    return processed_data

def do_data_processing(input_file):
    # Placeholder function that performs data processing
    processed_data = input_file
    return processed_data

def read_csv_file(file_path):
    with open(file_path, 'r') as file:
        reader = csv.reader(file)
        return list(reader)

def write_csv_file(file_path, data):
    with open(file_path, 'w') as file:
        writer = csv.writer(file)
        writer.writerows(data)

default_args = {
    'owner': 'airflow',
    'depends_on_past': False,
    'start_date': datetime(2021, 1, 1),
    'email_on_failure': False,
    'email_on_retry': False,
    'retries': 1,
    'retry_delay': timedelta(minutes=5),
}

dag = DAG(
    'data_processing_dag',
    default_args=default_args,
    description='A DAG that reads data from a CSV file, processes it'
                ', and writes it to a destination',
    schedule_interval=timedelta(hours=1),
)

read_file = PythonOperator(
    task_id='read_file',
    python_callable=lambda: read_csv_file('/path/to/input_file.csv'),
    xcom_push=True,
    dag=dag,
)

process_data = PythonOperator(
    task_id='process_data',
    python_callable=process_data,
    provide_context=True,
    dag=dag,
)

write_file = PythonOperator(
    task_id='write_file',
    python_callable=lambda: write_csv_file('/path/to/output_file.csv',
                                           ti.xcom_pull(task_ids='process_data')),
    provide_context=True,
    dag=dag,
)

read_file >> process_data >> write_file

Hay varios servicios de canalización de datos por lotes basados en la nube, como AWS Glue, Google Cloud Composer (basado en Airflow) y Azure Data Factory.

Una de las desventajas de Hadoop o de otros pipelines por lotes es la necesidad de leer datos del disco, procesarlos y escribirlos de nuevo en el disco en cada paso. Sin embargo, marcos como Spark y Dask saben cómo compilar el pipeline de procesamiento en un grafo óptimo en el que las tareas se realizan en memoria siempre que sea posible, lo que minimiza la IO a disco y maximiza el rendimiento.

El Ejemplo 4-6 muestra un código Spark que lee un archivo CSV, procesa los datos y escribe el resultado en un archivo de destino.

Ejemplo 4-6. Ejemplo de código de canalización de datos PySpark

from pyspark.sql import SparkSession

# Create a Spark session
spark = SparkSession.builder.appName("SimpleBatchProcessing").getOrCreate()

# Load a CSV file into a Spark DataFrame
df = spark.read.csv("/path/to/input_file.csv", header=True, inferSchema=True)

# Perform some data processing on the DataFrame
processed_df = df.groupBy("column_name").agg({"column_name": "mean"})

# Write the processed DataFrame to a new CSV file
processed_df.write.csv("/path/to/output_file.csv", header=True)

# Stop the Spark session
spark.stop()

El Ejemplo 4-7 muestra la misma tarea, implementada utilizando Dask. La ventaja de Dask es que las operaciones son muy similares a las de los pandas de Python, lo que supone una enorme ventaja para los científicos de datos. Sin embargo, Spark suele ser más escalable y robusto.

Ejemplo 4-7. Ejemplo de código de canalización de datos Dask

import dask.dataframe as dd

# Load a CSV file into a Dask DataFrame
df = dd.read_csv('/path/to/input_file.csv')

# Perform some data processing on the DataFrame
processed_df = df.groupby('column_name').column_name.mean().compute()

# Write the processed DataFrame to a new CSV file
processed_df.to_csv('/path/to/output_file.csv', index=False)

Puedes ver que los ejemplos de Spark y Dask son mucho más sencillos que los de Airflow. Sin embargo, Airflow puede ser más adecuado para gestionar y trazar trabajos largos y complejos.

Procesamiento de flujos

El procesamiento de flujos permite el procesamiento de datos escalable, tolerante a fallos y en tiempo real. Suele utilizarse en aplicaciones que procesan grandes cantidades de datos en tiempo real, como análisis en tiempo real, detección de fraudes o recomendaciones.

En el procesamiento de flujos, los datos y eventos entrantes se introducen en un flujo (cola) y son leídos por uno o más trabajadores. Los trabajadores procesan los datos secuencialmente, realizan transformaciones, agregan resultados y escriben los resultados en una base de datos o en un flujo de salida. A diferencia de las colas de mensajes tradicionales, el procesamiento del flujo se produce en orden. Por ejemplo, supongamos que el flujo contiene dos eventos: uno para el inicio de sesión del cliente y otro para el cierre de sesión del cliente. No procesarlos en orden puede conducir a un estado roto. Otro ejemplo es una operación de ingreso de dinero, seguida de una retirada. La retirada puede ser rechazada si las operaciones se procesan en el orden incorrecto.

Los flujos están diseñados para escalar. Se dividen en particiones, y cada partición maneja un conjunto específico de objetos de datos, por lo que no violará el orden. Por ejemplo, un flujo de actividad de usuario está particionado por el ID de usuario, de modo que las actividades de un usuario concreto siempre se almacenarán en la misma partición y serán procesadas por el mismo trabajador.

Los streams como Kafka, AWS Kinesis y otros son diferentes de las colas de mensajes como RabbitMQ, AMQP, Amazon SQS, Google Pub/Sub, etc. Las colas de mensajes no garantizan el orden de los mensajes. Sin embargo, garantizan una entrega fiable de los mensajes, mientras que el cliente gestiona la fiabilidad en el caso de los flujos. Además, son mucho más rápidas debido a la lógica más directa y al paralelismo que ofrecen los flujos.

La Figura 4-13 ilustra una aplicación de streaming en la que los clientes publican datos que se distribuyen entre las particiones individuales (basándose en un hash de la clave de partición). Un trabajador lee de cada partición y procesa los datos. El trabajador puede utilizar una base de datos para almacenar el estado en intervalos conocidos (puntos de control), de modo que el estado pueda recuperarse en caso de fallo, o el trabajador puede liberar la memoria no utilizada. Por último, los resultados pueden escribirse en una base de datos de destino o en un flujo de salida.

Los flujos proporcionan "semántica de al menos una vez". Por tanto, el mismo mensaje puede aparecer varias veces. Una forma de proporcionar una semántica "exactamente una vez" (el mismo mensaje se procesa una sola vez) es con la ayuda de los puntos de control. Los flujos se procesan en orden, y el estado puede persistir después de cada microlote. En caso de fallo, el trabajador puede restaurar los datos del último punto de control (estado), procesar los eventos a partir de ese punto e ignorar los eventos más antiguos.

Marcos de procesamiento de flujos

Hacer análisis en tiempo real sobre flujos en tiempo real difiere de hacerlo por lotes o SQL. Con los flujos, los trabajadores pueden repasar los datos una sola vez, en orden secuencial, y ver una parte de los datos (en la misma partición). Por eso los marcos de análisis en tiempo real, como Spark Streaming, Apache Flink, Apache Beam, Apache NiFi y otros, se centran en el procesamiento de flujos e implementan los métodos analíticos y estadísticos estándar de forma optimizada para los flujos.

Un escenario típico en el procesamiento de flujos es agregar valores a lo largo del tiempo; por ejemplo, examinar el valor total de las transacciones de los clientes en la última hora para detectar fraudes. No es factible calcular el total para cada nuevo evento con el procesamiento de flujos. Llevaría una cantidad considerable de tiempo y memoria. En su lugar, los valores se agrupan en cubos con ventanas, por ejemplo, seis cubos o más, cada uno de los cuales contiene el total por cada 10 minutos. El proceso suma los valores de sólo los últimos seis cubos y elimina el cubo más antiguo cada 10 minutos. La Figura 4-14 ilustra ventanas deslizantes superpuestas con una duración de ventana de un minuto y periodos de ventana de 30 segundos.

El Ejemplo 4-8 muestra el código de Apache Beam para definir una ventana de este tipo.

Ejemplo 4-8. Definir la ventana deslizante utilizando Apache Beam

from apache_beam import window
sliding_windowed_items = (
    items | 'window' >> beam.WindowInto(window.SlidingWindows(60, 30)))

Codificar con marcos de procesamiento de flujos requiere conocimientos avanzados de ingeniería de datos. Por eso muchos usuarios evitan los datos en tiempo real, a pesar de que pueden proporcionar un valor empresarial mucho mejor y unos resultados de puntuación de modelos más precisos. Los almacenes de características vienen al rescate, ya que pueden generar automáticamente la canalización por lotes y la canalización de flujo a partir de la misma lógica de procesamiento de datos de nivel superior .

Tiendas especializadas

Almacenes de características son una fábrica y un repositorio central para las características del aprendizaje automático. Los almacenes de características gestionan la recopilación de datos brutos de diversas fuentes, el proceso de transformación, el almacenamiento, la catalogación, el control de versiones, la seguridad, el servicio y el monitoreo. Automatizan muchos procesos descritos en este capítulo, al tiempo que aceleran el tiempo de producción y reducen los esfuerzos de ingeniería. Los almacenes de características forman un catálogo compartido de características listas para la producción, permiten la colaboración y el intercambio entre equipos, y aceleran la innovación y la entrega de nuevas aplicaciones de IA.

Las primeras implementaciones de almacenes de características procedían de grandes proveedores de servicios como Uber, Twitter y Spotify. En esos proveedores, la IA es esencial para el negocio, y los almacenes de características les ayudaron a acelerar el desarrollo y la implementación de nuevas aplicaciones de IA y a mejorar la colaboración y la reutilización. Hoy existen múltiples implementaciones comerciales y de código abierto entre las que elegir.

Los almacenes de funciones avanzadas proporcionan las siguientes capacidades:

Conectividad de datos: Integración sin cola con múltiples fuentes offline (lagos de datos, almacenes de datos, bases de datos, etc.) y online (flujos, colas de mensajes, API, servicios gestionados, etc.).
Transformación offline y online: Algunos almacenes de funciones ofrecen capacidades para construir y gestionar automáticamente los pipelines de lotes y streaming a partir de la lógica de nivel superior.
Almacenamiento: Almacenar las características generadas en un almacén fuera de línea (como un almacén de objetos) y en un almacén en línea (normalmente una base de datos clave/valor).
Gestión de metadatos: Autogeneración, almacenamiento y gestión de todos los metadatos de características, incluyendo linaje, esquemas, estadísticas, etiquetas y más.
Versionado: La gestión de múltiples versiones de cada característica y el proceso de promoción de características del desarrollo a la producción y la integración con CI/CD.
Generar y gestionar vectores de características: Unir correctamente varias características en un único conjunto de datos para utilizarlo en aplicaciones de entrenamiento o de servicio.
Catalogación central: Proporcionar acceso centralizado para generar, etiquetar o buscar funciones.
Seguridad y gobernanza: Controlar el acceso a las funciones y a los datos brutos y registrar el acceso a las funciones.
Interfaz de usuario y SDK fáciles de usar: Acceso sencillo a través de API y una interfaz de usuario para abstraer la complejidad del subrayado, visualizar las características y hacer que los científicos de datos puedan utilizarlo.
Monitoreo y alta disponibilidad: Monitorea los activos y las tareas de procesamiento de datos de forma automática, al tiempo que se recupera de fallos de forma fiable.
Validación y análisis de características: Ejecutar diversas tareas de procesamiento de datos de forma automática o iniciadas por el usuario, para validar la corrección de los rasgos o generar un análisis profundo de los rasgos, la correlación, etc.

Debes comparar a fondo las capacidades antes de elegir un almacén de funciones. Por ejemplo, muchos tienen una funcionalidad muy parcial, pueden centrarse en catalogar características o carecer de transformaciones automatizadas, gestión de datos a escala y funcionalidad en tiempo real. Estas capacidades aportan el valor más significativo a la hora de acelerar el tiempo de producción.

Arquitectura y uso del almacén de prestaciones

La Figura 4-15 ilustra la arquitectura general y el uso de un almacén de características. Los datos brutos se ingieren y se transforman en características, y las características se catalogan y se sirven a diferentes aplicaciones (entrenamiento, servicio, monitoreo). Las API y una interfaz de usuario permiten a los científicos de datos, ingenieros de datos e ingenieros de ML actualizar, buscar, monitorizar y utilizar las características.

Los componentes básicos de un almacén de características son

Capa de transformación: Convierte los datos brutos offline u online en características y los almacena tanto en un almacén online (clave/valor) como offline (objeto).
Capa de almacenamiento: Almacena varias versiones de una característica en tablas de características (conjuntos de características) y gestiona el ciclo de vida de los datos (crear, añadir, eliminar, monitorizar y asegurar los datos). La capa de datos almacena cada característica de dos formas: fuera de línea para la formación y el análisis, y en línea para el servicio y el monitoreo.
Recuperación de características: Acepta solicitudes de múltiples características (vectores de características) y otras propiedades (como intervalos de tiempo y datos de eventos), y produce una instantánea de datos fuera de línea para entrenamiento o un vector en tiempo real para servir.
Gestión de metadatos y catalogación: Almacena la definición del rasgo, los metadatos, las etiquetas y las relaciones.

Servicio de Ingestión y Transformación

En este capítulo se han tratado las complejidades de implementar el procesamiento a gran escala de datos por lotes y en tiempo real, el versionado de datos y la gestión de metadatos. Los almacenes de características pretenden reducir esa complejidad mediante la abstracción y la automatización. Con los almacenes de características modernos, los conductos de datos se describen mediante una lógica de transformación de alto nivel. Esta lógica se convierte a la semántica del motor de procesamiento subyacente y se implementa como un servicio continuo y de nivel de producción, ahorrando importantes esfuerzos de ingeniería.

La implementación de canalizaciones es diferente para el desarrollo local (utilizando paquetes como pandas), los datos fuera de línea a gran escala (utilizando el procesamiento por lotes) y los datos en tiempo real (utilizando el procesamiento por flujos). La ventaja de un almacén de características que admite transformaciones automatizadas es que utiliza una definición para los tres modos de implementación y elimina la reingeniería que supone portar canalizaciones de datos de un método a otro. En algunos almacenes de características, la tecnología de canalización de datos vendrá determinada por las fuentes de datos, ya sean offline (lagos de datos, almacenes de datos, bases de datos, etc.) u online (flujos, colas de mensajes, API, servicios gestionados y otros).

Los almacenes de características implementan la ingesta y transformación de datos en grupos de características (llamados conjuntos de características o grupos de características) que proceden de la misma fuente; por ejemplo, todas las características extraídas del registro de transacciones de una tarjeta de crédito. Los conjuntos de características toman datos de fuentes offline u online, construyen una lista de características mediante un conjunto de transformaciones, y almacenan las características resultantes junto con los metadatos y estadísticas asociados.

La Figura 4-16 ilustra el servicio de transformación (conjunto de características). Una vez que los datos se ingieren desde la fuente, pasan a través de un grafo (DAG) de transformaciones, y las características resultantes se escriben en los almacenes offline y online.

Ejemplos de transformación (por tipo de datos):

Estructurado: Filtrar, agrupar, unir, agregar, codificación OneHot, mapear, extraer y clasificar
Textual: Extraer, analizar, desensamblar, detectar entidades, sentimientos e incrustaciones
Visual (imágenes y vídeos): Encuadrar, redimensionar, detectar objetos, recortar, recolorear, girar, mapear y clasificar

El servicio de transformación generado debe ser de nivel de producción y admitir autoescalado, alta disponibilidad, actualizaciones en vivo, etc. Además, debe admitir la ingesta y el procesamiento continuos de datos. Por ejemplo, los nuevos datos pueden llegar continuamente (en tiempo real) o en intervalos programados (fuera de línea). Por lo tanto, las tecnologías de funciones sin servidor son un ajuste excelente.

Función Almacenamiento

Las funciones suelen almacenarse de dos formas: almacenamiento fuera de línea para aplicaciones de formación y análisis, y almacenamiento en línea para aplicaciones de servicio y monitoreo en tiempo real. Véase la Figura 4-17.

El almacén offline contiene todos los datos históricos y a menudo utiliza lagos de datos, almacenamiento de objetos o tecnologías de almacén de datos. Por ejemplo, una opción habitual es utilizar archivos Parquet comprimidos almacenados en un almacenamiento de objetos como AWS S3.

El almacén en línea contiene los datos más recientes y a menudo utiliza almacenes NoSQL o de clave/valor como Redis, AWS DynamoDB, Google BigTable y otros. El almacén en línea tiene que soportar funciones de lectura en milisegundos.

Recuperación de características (para entrenar y servir)

Las aplicaciones de entrenamiento, servicio y análisis requieren múltiples características de múltiples conjuntos de datos y fuentes. En cambio, los almacenes de características organizan las características en grupos (denominados conjuntos de características) en función de su origen y entidad (clave primaria, como un id de usuario, id de producto, etc.).

Recuperar múltiples características de diferentes fuentes, momentos y con diferentes índices puede ser una tarea analítica compleja. Los almacenes de características determinan automáticamente los parámetros necesarios para la consulta JOIN basándose en los metadatos de las características, los nombres de las entidades y los datos de la solicitud del usuario. Además, cuando los conjuntos de datos son transaccionales (los registros están marcados con una marca de tiempo), la operación de unión debe tener en cuenta la corrección temporal y el desplazamiento temporal para devolver sólo los valores conocidos en el momento del evento (también denominada operación analítica as of join ).

Los conjuntos de características fuera de línea pueden generarse mediante consultas SQL generadas por el almacén de características. Sin embargo, con aplicaciones de servicio en tiempo real que necesitan responder en milisegundos, esto crea una sobrecarga considerable, y se utilizan otros métodos en tiempo real. Además, las características basadas en el tiempo (como el número de solicitudes en la última hora) no pueden calcularse previamente y requieren un manejo especial para generar un resultado exacto (por ejemplo, combinando datos precalculados con ventana de tiempo y cálculos ad hoc de última milla).

La Figura 4-18 ilustra el flujo de recuperación de características con dos motores separados, uno para la recuperación fuera de línea y otro para la recuperación en tiempo real. Observa que, en el caso de la recuperación fuera de línea, el conjunto de datos se instantanea o se conserva en un nuevo conjunto de datos para permitir el seguimiento del linaje de los datos y su explicabilidad.

La petición get_offline_features puede aceptar datos de eventos en los que basar la consulta, un intervalo de tiempo válido (por ejemplo, si queremos entrenar el modelo basándonos en datos del último mes), y qué características y columnas debe devolver (por ejemplo, si incluir las columnas de índice, tiempo o etiqueta). A continuación, inicia un trabajo de análisis local o sin servidor que calcule los resultados y devuelva el conjunto de datos del vector de características.

En la recuperación en tiempo real, el sistema inicializa el servicio de recuperación (configurando una función de análisis en tiempo real local o remota una vez para ahorrar tiempo en las peticiones). Después, las peticiones de los usuarios se envían con las claves de entidad (tomadas de los datos del evento) y aceptan un vector de resultados. Además, algunos almacenes de características permiten la imputación en tiempo real (sustituyendo los datos perdidos o NaN por valores estadísticos de características tomados de los metadatos de características).

Soluciones y ejemplos de uso de las Tiendas de Características

Los almacenes de características empezaron como plataformas internas en los principales proveedores de servicios en la nube (como Uber, Spotify y Twitter). Pero ahora, existen en el mercado muchas soluciones de feature store comerciales y de código abierto. Sin embargo, como en todo nuevo espacio tecnológico importante, hay muchas diferencias de funcionalidad entre esas soluciones; debes ser consciente de ello para poder elegir la adecuada.

La diferencia más notable y esencial es si la plataforma de almacén de características gestiona la canalización de datos (transformación) por ti y si admite canalizaciones tanto fuera de línea como en tiempo real (streaming). Como has leído en este capítulo, construir y gestionar una canalización de datos escalable es el mayor reto. Si te ves obligado a hacerlo manualmente, se socava considerablemente el valor de un almacén de características.

La Tabla 4-2 compara las principales soluciones de almacén de características:

Tabla 4-2. Comparación de soluciones de almacén de características
Categoría	Fiesta	Tecton	MLRun	SageMaker	Vértice AI	Databricks	HopsWorks
Código abierto	Sí	No	Sí	No	No	No	Sí
Opción gestionada	No	nubes mayores	nube + on-prem	en AWS	en GCP	nubes mayores	nube + on-prem
Canalizaciones offline	No	Sí	Sí	No	No	No	Sí
Canalizaciones en tiempo real	No	Sí	Sí	No	No	No	No
Recuperación de características	Sí	Sí	Sí	Sí	Sí	Sí	Sí
Motores	Chispa	Chispa	Python, Dask, Spark, Nuclio	Ninguno	Chispa	Chispa	Spark, Flink
Análisis de funciones	No	Sí	Sí	No	No	No	Sí
Versionado y linaje	No	Sí	Sí	No	No	No	Sí
Características seguridad	No	Sí	Sí	Sí	No	No	No
Monitoreo	No	Sí	Sí	No	No	No	Sí
Entrenar y servir sin pegamento	No	No	Sí	No	No	No	Sí

Las siguientes secciones demostrarán cómo se utilizan los almacenes de características con los dos principales marcos de código abierto: Feast y MLRun. Ten en cuenta que MLRun es más completo y proporciona servicios de transformación offline y online (basados en los motores sin servidor de MLRun) junto con muchas otras características únicas.

Utilizar la tienda de funciones de Festín

Feast no proporciona un servicio de transformación. Los datos deben prepararse por adelantado y almacenarse en una fuente compatible (como S3, GCS, BigQuery). Feast registra el conjunto de datos fuente y sus metadatos (esquema, entidad, etc.) en un objeto FeatureView, como se muestra en el Ejemplo 4-9.

Ejemplo 4-9. Definición de Feast FeatureView (fuente: Feast)

# Read data from parquet files. Parquet is convenient for local development mode.
# For production, you can use your favorite DWH, such as BigQuery. See Feast
# documentation for more info.
driver_hourly_stats = FileSource(
    name="driver_hourly_stats_source",
    path="/content/feature_repo/data/driver_stats.parquet",
    timestamp_field="event_timestamp",
    created_timestamp_column="created",
)

# Define an entity for the driver. You can think of entity as a primary key used to
# fetch features.
driver = Entity(name="driver", join_keys=["driver_id"])

# Our parquet files contain sample data that includes a driver_id column, timestamps
# and three feature column. Here we define a Feature View that will allow us to serve
# this data to our model online.
driver_hourly_stats_view = FeatureView(
    name="driver_hourly_stats",
    entities=[driver],
    ttl=timedelta(days=1),
    schema=[
        Field(name="conv_rate", dtype=Float32),
        Field(name="acc_rate", dtype=Float32),
        Field(name="avg_daily_trips", dtype=Int64),
    ],
    online=True,
    source=driver_hourly_stats,
    tags={},
)

Feast no proporciona un servicio de transformación o ingestión en línea. En su lugar, el usuario necesita ejecutar una tarea de materialización para copiar las características offline en el almacén en tiempo real (base de datos). Por desgracia, esto también significa que los datos almacenados en el almacén en línea son imprecisos entre materializaciones, y ejecutar la materialización con demasiada frecuencia puede dar lugar a una sobrecarga computacional significativa.

Ejecutar la tarea de materialización a través del SDK:

store = FeatureStore(repo_path=".")
store.materialize_incremental(datetime.now())

El proyecto puede contener una o varias vistas de características, y cada una se define y materializa independientemente. Las características pueden recuperarse de una o varias vistas de características (se iniciará una operación JOIN).

Para recuperar funciones offline (directamente de la fuente offline), utiliza la llamada a la API get_historical_features() como se muestra en el Ejemplo 4-10.

Ejemplo 4-10. Recuperar funciones sin conexión con Feast (fuente: Feast)

# The entity dataframe is the dataframe we want to enrich with feature values
# see https://docs.feast.dev/getting-started/concepts/feature-retrieval for details
# for all entities in the offline store instead
entity_df = pd.DataFrame.from_dict(
    {
        # entity's join key -> entity values
        "driver_id": [1001, 1002, 1003],
        # "event_timestamp" (reserved key) -> timestamps
        "event_timestamp": [
            datetime(2021, 4, 12, 10, 59, 42),
            datetime(2021, 4, 12, 8, 12, 10),
            datetime(2021, 4, 12, 16, 40, 26),
        ],
        # (optional) label name -> label values. Feast does not process these
        "label_driver_reported_satisfaction": [1, 5, 3],
        # values we're using for an on-demand transformation
        "val_to_add": [1, 2, 3],
        "val_to_add_2": [10, 20, 30],
    }
)

store = FeatureStore(repo_path=".")

# retrieve offline features, feature names are specified with <view>:<feature-name>
training_df = store.get_historical_features(
    entity_df=entity_df,
    features=[
        "driver_hourly_stats:conv_rate",
        "driver_hourly_stats:acc_rate",
        "driver_hourly_stats:avg_daily_trips",
        "transformed_conv_rate:conv_rate_plus_val1",
        "transformed_conv_rate:conv_rate_plus_val2",
    ],
).to_df()

print("----- Example features -----\n")
print(training_df.head())

Para recuperar funciones online de la tienda online, utilizamos la llamada a la API get_online_features(), como se muestra en el Ejemplo 4-11.

Ejemplo 4-11. Recuperar funciones en línea con Festín (fuente: Festín)

from pprint import pprint
from feast import FeatureStore

store = FeatureStore(repo_path=".")

feature_vector = store.get_online_features(
    features=[
        "driver_hourly_stats:acc_rate",
        "driver_hourly_stats:avg_daily_trips",
        "transformed_conv_rate:conv_rate_plus_val1",
        "transformed_conv_rate:conv_rate_plus_val2",
    ],
    entity_rows=[
        # {join_key: entity_value}
        {
            "driver_id": 1001,
            "val_to_add": 1000,
            "val_to_add_2": 2000,
        },
        {
            "driver_id": 1002,
            "val_to_add": 1001,
            "val_to_add_2": 2002,
        },
    ],
).to_dict()

pprint(feature_vector)

# results:
{'acc_rate': [0.86463862657547, 0.6959823369979858],
 'avg_daily_trips': [359, 311],
 'conv_rate_plus_val1': [1000.6638441681862, 1001.1511893719435],
 'conv_rate_plus_val2': [2000.6638441681862, 2002.1511893719435],
 'driver_id': [1001, 1002]}

Utilizar el almacén de funciones de MLRun

MLRun admite el registro de fuentes existentes (como Feast) o la definición de una canalización de datos para transformar los datos de la fuente en características. Al definir el conducto de datos (llamado grafo), MLRun aprovisiona el motor de procesamiento de datos seleccionado basándose en las definiciones abstractas del usuario. MLRun admite algunos motores de procesamiento, como Python local, Dask, Spark y Nuclio (un motor sin servidor en tiempo real).

En MLRun, por defecto, el pipeline escribe en almacenes online y offline, por lo que no hay necesidad de trabajos de materialización separados, y las funciones online y offline están siempre sincronizadas. Además, MLRun puede autodetectar el esquema de datos, haciéndolo más sencillo y robusto.

MLRun separa la definición del conjunto de características (una colección de características generadas por la misma canalización) de las definiciones de la fuente de datos. De este modo, puedes utilizar el mismo conjunto de características en desarrollo interactivo y en producción. Sólo tienes que cambiar la fuente de un archivo local en desarrollo a una base de datos o un flujo Kafka en tiempo real en la implementación en producción.

El Ejemplo 4-12 muestra un ejemplo de definición de un conjunto de características para procesar transacciones de tarjetas de crédito con el fin de detectar fraudes con tarjetas de crédito. La definición incluye la entidad, la marca de tiempo y el gráfico de transformación utilizando operadores y agregaciones incorporados. Ten en cuenta que un usuario también puede añadir sus operadores Python personalizados. Ver el ejemplo completo.

La canalización de datos consiste en lo siguiente

Extraer los componentes de los datos (hora, día de la semana).
Asignación de los valores de edad
Codificación de una sola vez para la categoría de transacción y el género
Agregando la cantidad (avg, sum, count, max sobre ventanas de tiempo de 2/12/24 horas)
Agregando las transacciones por categoría (en ventanas temporales de 14 días)
Escribir los resultados en objetivos offline (Parquet) y online (NoSQL)

Ejemplo 4-12. Definición del FeatureSet de MLRun (fuente: MLRun)

import mlrun.feature_store as fs

# Define the credit transactions FeatureSet
transaction_set = fs.FeatureSet("transactions",
                                entities=[fs.Entity("source")],
                                timestamp_key='timestamp',
                                description="transactions feature set")

# Define and add value mapping
main_categories = ["es_transportation", "es_health", "es_otherservices",
       "es_food", "es_hotelservices", "es_barsandrestaurants",
       "es_tech", "es_sportsandtoys", "es_wellnessandbeauty",
       "es_hyper", "es_fashion", "es_home", "es_contents",
       "es_travel", "es_leisure"]

# One Hot Encode the newly defined mappings
one_hot_encoder_mapping = {'category': main_categories,
                           'gender': list(transactions_data.gender.unique())}

# Define the data pipeline (graph) steps
transaction_set.graph\
    .to(DateExtractor(parts = ['hour', 'day_of_week'],
        timestamp_col = 'timestamp'))\
    .to(MapValues(mapping={'age': {'U': '0'}}, with_original_features=True))\
    .to(OneHotEncoder(mapping=one_hot_encoder_mapping))


# Add aggregations for 2, 12, and 24 hour time windows
transaction_set.add_aggregation(name='amount',
                                column='amount',
                                operations=['avg','sum', 'count','max'],
                                windows=['2h', '12h', '24h'],
                                period='1h')


# Add the category aggregations over a 14 day window
for category in main_categories:
    transaction_set.add_aggregation(name=category,column=f'category_{category}',
                                    operations=['count'], windows=['14d'],
                                        period='1d')

El canal de datos puede visualizarse utilizando transaction_set.plot(rankdir="LR", with_targets=True), como se ve en la Figura 4-19.

Una vez que tengas la definición del conjunto de características, puedes probarlo y depurarlo con el métodopreview() que ejecuta el canal de datos localmente y te permite ver los resultados:

df = fs.preview(transaction_set, transactions_data)
df.head()

Una vez definido el conjunto de funciones, puedes implementarlo como un trabajo de producción que se ejecuta bajo demanda, en un horario determinado o como una canalización en tiempo real.

Para ejecutar la ingesta por lotes, utiliza el método ingest(). Para la ingesta en tiempo real desde HTTP o flujos, utiliza deploy_ingestion_service_v2(), que inicia un canal sin servidor Nuclio en tiempo real. Mira el Ejemplo 4-13.

Ejemplo 4-13. Ingesta de datos en el FeatureSet de MLRun (fuente: MLRun)

# Batch ingest the transactions dataset (from CSV file) through the defined pipeline
source = CSVSource("mycsv", path="measurements.csv")
fs.ingest(transaction_set, source=source)

# Deploy a real-time pipeline with HTTP API endpoint as the source
# MLRun support other real-time sources like Kafka, Kinesis, etc.
source = HTTPSource()
fs.deploy_ingestion_service_v2(transaction_set, source)

Puedes ver los conjuntos de características, sus metadatos y estadísticas en la interfaz de usuario del almacén de características MLRun. Mira la Figura 4-20.

La recuperación de características en MLRun se realiza mediante el objeto vector de características. Los vectores de características contienen las definiciones de las características solicitadas y parámetros adicionales. Además, también almacenan valores calculados como los metadatos de las características, estadísticas, etc., que pueden ser útiles en tareas de entrenamiento, servicio o monitoreo. Por ejemplo, las estadísticas de las características se utilizan para la imputación automática de valores en caso de que falten valores de características o sean NaN, y para el monitoreo de la deriva del modelo en la aplicación de servicio.

Los vectores de características pueden crearse, actualizarse y visualizarse en la interfaz de usuario de MLRun.

Los usuarios definen primero el vector de características, y luego pueden utilizarlo para obtener características offline u online. Mira cómo obtener características offline y utilizar el método get_offline_features() en el Ejemplo 4-14.

Ejemplo 4-14. Obtener funciones offline de MLRun (fuente: MLRun)

# Define the list of features you will be using (<feature-set>.<feature>)
features = ['transactions.amount_max_2h',
            'transactions.amount_sum_2h',
            'transactions.amount_count_2h',
            'transactions.amount_avg_2h',
            'transactions.amount_max_12h']

# Import MLRun's Feature Store
import mlrun.feature_store as fstore

# Define the feature vector name for future reference
fv_name = 'transactions-fraud'

# Define the feature vector using our Feature Store
transactions_fv = fstore.FeatureVector(fv_name, features,
                                   label_feature="labels.label",
                                   description=
                                       'Predicting a fraudulent transaction')

# Save the feature vector definition in the Feature Store
transactions_fv.save()

# Get offline feature vector as dataframe and save the dataset to a parquet file
train_dataset = fstore.get_offline_features(transactions_fv, target=ParquetTarget())

# Preview the dataset
train_dataset.to_dataframe().tail(5)

Para obtener características en tiempo real, primero tienes que definir un servicio (que inicializa la tubería de recuperación en tiempo real), seguido de los métodos .get() para solicitar valores de características en tiempo real. La separación entre la creación del servicio (inicialización única) y las solicitudes individuales garantiza una menor latencia de las solicitudes. Además, MLRun admite la imputación automática de valores basada en los metadatos y estadísticas de la característica. Esto puede ahorrar una importante sobrecarga de desarrollo y cálculo. Véase el Ejemplo 4-15.

Ejemplo 4-15. Obtener funciones online de MLRun (fuente: MLRun)

# Create the online feature service, substitute NaN values with
# the feature mean value
svc = fstore.get_online_feature_service('transactions-fraud:latest',
                                    impute_policy={"*": "$mean"})

# Get sample feature vector
sample_fv = svc.get([{'source': 'C76780537'}])

# sample_fv Result
[{'amount_max_2h': 14.68,
  'amount_max_12h': 70.81,
  'amount_sum_2h': 14.68,
  'amount_count_2h': 1.0,
  'amount_avg_2h': 14.68}]

Nota

Los almacenes de características de MLRun proporcionan agregaciones precisas en tiempo real y baja latencia combinando valores precalculados durante el proceso de ingestión con cálculos en tiempo real en el momento de solicitar las características.

El marco MLRun proporciona una canalización de desarrollo y entrenamiento de modelos, canalizaciones de servicio en tiempo real y un monitoreo integrado de modelos. El almacén de características de MLRun se integra de forma nativa con los demás componentes, eliminando la lógica de cola redundante, la traducción de metadatos, etc., lo que acelera el tiempo de producción .

Conclusión

Dado que la gestión y el procesamiento de datos son los componentes más críticos del ML, es importante comprender cómo realizar de forma óptima las tareas relacionadas con los datos. Este capítulo explora las herramientas y prácticas recomendadas para las distintas etapas del trabajo con tus datos. Comenzamos el capítulo hablando del versionado y el linaje de los datos, que son esenciales para rastrear el origen de los datos. Luego exploramos la preparación y el análisis de datos a escala, que es cómo se manejan los datos para poder utilizarlos en producción. En esta sección, también hablamos de la arquitectura de las soluciones interactivas de procesamiento de datos y de las diferencias entre el procesamiento de datos por lotes y el procesamiento en tiempo real.

Tras repasar los retos de aplicar estas prácticas a escala, pasamos a presentar el concepto de almacenes de características, que son un repositorio central de características de ML. Cubrimos las capacidades de un almacén de características, como la conectividad de datos y la transformación offline y online. También mostramos dónde encaja el almacén de características en el proceso de MLOps, desde la ingesta de datos sin procesar hasta el apoyo al uso de esos datos en el entrenamiento, el servicio, el monitoreo y más. Por último, revisamos distintas soluciones de almacén de características y cómo utilizarlas.

Preguntas de debate sobre el pensamiento crítico

¿Qué detalles proporcionan los metadatos? Como profesionales de los datos, ¿por qué necesitamos esta información?
¿Qué herramientas de versionado de datos de código abierto hay disponibles? ¿Cuál podría ser adecuada para tu organización?
¿Cuál es la diferencia entre el procesamiento por lotes y el procesamiento por flujos? ¿Cuándo se utiliza cada uno?
¿Cómo simplifica un almacén de características las prácticas de gestión y procesamiento de datos? ¿Qué capacidades lo permiten?
¿Cuáles son las diferencias entre las tiendas de funciones Feast y MLRun? ¿Cuál podría ser una buena opción para tu organización?

Ejercicios

Elige una solución de código abierto (DVC, Pachyderm, MLflow o MLRun) y crea un guión o flujo de trabajo de versionado de datos que registre y versione los datos y metadatos.
Crea un prototipo de una tubería de procesamiento por lotes con la herramienta que elijas.
Conecta un conector de datos Trino a una fuente de datos.
Entrena un modelo de demostración (puedes utilizar Cara Abrazada si necesitas un modelo de muestra) con un almacén de características.
Crea un conjunto de características y un canal de ingestión en MLRun. Puedes utilizar este proyecto como referencia.

Get Implantar MLOps en la empresa now with the O’Reilly learning platform.

O’Reilly members experience books, live events, courses curated by job role, and more from O’Reilly and nearly 200 top publishers.

Start your free trial

Capítulo 4. Trabajar con almacenes de datos y funciones

Versionado y linaje de datos

Cómo funciona

Figura 4-1. Flujo de datos

Figura 4-2. Línea de datos en una canalización multietapa

Figura 4-3. Cómo afectan los cambios de datos, parámetros y código a las versiones de los artefactos

Herramientas comunes de versionado de datos de ML

Control de versiones de datos

Figura 4-4. Flujo de datos DVC (fuente: DVC)

Ejemplo 4-1. Añadir un paso de flujo de trabajo en DVC

Paquidermo

Ejemplo 4-2. Ejemplo de tubería Pachyderm

Figura 4-5. Interfaz de usuario de Pachyderm

Seguimiento MLflow

Ejemplo 4-3. Ejemplo de código de Seguimiento MLflow

Figura 4-6. Interfaz de usuario de MLflow

MLRun

Figura 4-7. Interfaz de usuario de ejecución de trabajos MLRun

Nota

Ejemplo 4-4. Ejemplo de código MLRun

Figura 4-8. Ver un artefacto de conjunto de datos en MLRun (con vista previa autogenerada, esquema y estadísticas)

Figura 4-9. Visualizar un artefacto gráfico interactivo utilizando el SDK de MLRun (en Jupyter)

Otros marcos

Preparación y análisis de datos a escala

Figura 4-10. Recuento de palabras distribuido con arquitectura map-reduce (fuente: O'Reilly)

Transformaciones de datos estructurados y no estructurados

Arquitecturas Distribuidas de Proceso de Datos

Tratamiento interactivo de datos

Figura 4-11. Arquitectura de PrestoDB y Trino (fuente: Presto)

Procesamiento de datos por lotes

Figura 4-12. Interfaz de usuario del flujo de aire

Ejemplo 4-5. Ejemplo de código de canalización de datos de flujo de aire

Ejemplo 4-6. Ejemplo de código de canalización de datos PySpark

Ejemplo 4-7. Ejemplo de código de canalización de datos Dask

Procesamiento de flujos

Figura 4-13. Arquitectura de la aplicación de streaming

Marcos de procesamiento de flujos

Figura 4-14. Ventanas deslizantes (fuente: Apache Beam)

Ejemplo 4-8. Definir la ventana deslizante utilizando Apache Beam

Tiendas especializadas

Arquitectura y uso del almacén de prestaciones

Figura 4-15. Uso y arquitectura del almacén de características

Servicio de Ingestión y Transformación

Figura 4-16. Ejemplo de canalización del servicio de transformación de características (conjunto de características)

Función Almacenamiento

Figura 4-17. Almacenamiento de características

Recuperación de características (para entrenar y servir)

Figura 4-18. Recuperación de características

Soluciones y ejemplos de uso de las Tiendas de Características

Utilizar la tienda de funciones de Festín

Ejemplo 4-9. Definición de Feast FeatureView (fuente: Feast)

Ejemplo 4-10. Recuperar funciones sin conexión con Feast (fuente: Feast)

Ejemplo 4-11. Recuperar funciones en línea con Festín (fuente: Festín)

Utilizar el almacén de funciones de MLRun

Ejemplo 4-12. Definición del FeatureSet de MLRun (fuente: MLRun)

Figura 4-19. Gráfico del conjunto de características

Ejemplo 4-13. Ingesta de datos en el FeatureSet de MLRun (fuente: MLRun)

Figura 4-20. FeatureSet de MLRun en la interfaz de usuario

Ejemplo 4-14. Obtener funciones offline de MLRun (fuente: MLRun)

Ejemplo 4-15. Obtener funciones online de MLRun (fuente: MLRun)

Nota

Conclusión

Preguntas de debate sobre el pensamiento crítico

Ejercicios

Don’t leave empty-handed

It’s yours, free.

Check it out now on O’Reilly