Data Pipelines de MySQL a HDFS

Proceso para mover y gestionar datos de una Base de Datos relacional a un Sistema de Archivo Distribuido HDFS con la ayuda de la ingesta de datos de SQOOP.

1.1 ¿Qué es Big Data?

Big Data se refiere a conjuntos de datos extremadamente grandes y complejos que no pueden ser gestionados o procesados con herramientas tradicionales. Su importancia radica en su capacidad para ofrecer insights valiosos y tomar decisiones basadas en datos.

1.1.1 Las 5 V’s de Big Data

1. Volumen: La cantidad masiva de datos generados a diario por fuentes como redes sociales, sensores IoT, y transacciones en línea.
2. Velocidad: La rapidez con la que se generan y procesan los datos en tiempo real o casi real.
3. Variedad: La diversidad de formatos y fuentes de datos, desde texto y videos hasta datos estructurados y no estructurados.
4. Veracidad: La calidad y confiabilidad de los datos, que puede variar dependiendo de la fuente.
5. Valor: El beneficio tangible que se obtiene al analizar y utilizar estos datos.

1.1.2 Ejemplos de Big Data en acción

• Plataformas de streaming que recomiendan contenido personalizado.
• Sistemas de transporte que optimizan rutas basadas en datos de tráfico en tiempo real.
• Aplicaciones de salud que analizan datos de pacientes para diagnósticos más precisos.

1.2 Arquitectura de Big Data: Cómo Funciona el Corazón de la Gestión de Grandes Datos

La arquitectura de Big Data es el marco que permite gestionar y procesar grandes volúmenes de datos de manera eficiente. Sin una infraestructura adecuada, sería casi imposible manejar la velocidad y la magnitud de los datos generados en la actualidad. A continuación, exploramos los componentes clave que permiten que Big Data funcione de manera efectiva.

1.2.1 Capa de Ingesta de Datos: La Puerta de Entrada

La ingesta de datos es el primer paso en la arquitectura de Big Data. Se trata de captar datos provenientes de múltiples fuentes: bases de datos, sensores IoT, redes sociales, aplicaciones web y más. Los datos pueden estar en diversos formatos, desde texto hasta imágenes o videos, y deben ser procesados para su análisis posterior.

Herramientas populares:

• Apache Kafka: Maneja flujos de datos en tiempo real.
• Apache Flume: Captura grandes cantidades de datos desde diversas fuentes.
• Apache Sqoop: Se utiliza para mover datos entre bases de datos tradicionales (como MySQL, Oracle, PostgreSQL) y plataformas de Big Data. Es particularmente útil para integrar bases de datos SQL con Hadoop y realizar migraciones masivas de datos.

1.2.2 Capa de Almacenamiento: El Gran Depósito de Datos

Una vez que los datos son recolectados, necesitan ser almacenados de manera distribuida. Aquí es donde entran los sistemas de almacenamiento de Big Data, diseñados para manejar grandes volúmenes de datos y ofrecer acceso rápido cuando sea necesario.

Componentes clave:

• Hadoop Distributed File System (HDFS): Un sistema de archivos distribuido que divide los datos en bloques, lo que facilita su almacenamiento y acceso de manera eficiente.
• Bases de datos NoSQL: Herramientas como Apache HBase y Cassandra que permiten almacenar datos no estructurados y escalarlos de manera flexible.

1.2.3. Capa de Procesamiento: Transformando los Datos Brutos

El procesamiento de datos es donde la magia realmente ocurre. Aquí, los datos almacenados se procesan y transforman para ser útiles en análisis o aplicaciones en tiempo real. Dependiendo de las necesidades, esto puede ser en lotes (batch) o en tiempo real.

Herramientas de procesamiento:

• MapReduce: Un modelo de programación que permite procesar grandes volúmenes de datos de manera paralela y distribuida.
• Apache Spark: Mucho más rápido que MapReduce, Spark permite procesamiento en memoria y es ideal tanto para procesamiento batch como en tiempo real.

1.2.4. Capa de Análisis: Generando Insights

Una vez que los datos han sido procesados, es el momento de analizar y extraer valor. Aquí es donde entran en juego los algoritmos de análisis, los modelos predictivos y el aprendizaje automático (Machine Learning). Esta capa es crucial para la toma de decisiones basada en datos.

Herramientas de análisis:

• Apache Hive: Permite realizar consultas SQL sobre datos almacenados en Hadoop.
• Apache Pig: Ideal para procesar y analizar grandes cantidades de datos de manera más sencilla que con MapReduce.

1.2.5. Capa de Visualización: Haciendo los Datos Comprensibles

Los datos analizados deben ser presentados de manera clara y comprensible. Aquí entra la visualización de datos, que permite representar los resultados de manera gráfica, ya sea a través de gráficos, dashboards o mapas interactivos.

Herramientas de visualización:

• Tableau, Power BI y Qlik son algunas de las herramientas que permiten crear visualizaciones interactivas y fáciles de entender.

1.2.6. Capa de Seguridad y Gobernanza: Protegiendo los Datos

Con tantos datos en juego, la seguridad y la gobernanza son cruciales. Es importante garantizar que los datos estén protegidos y que solo las personas autorizadas tengan acceso a ellos. Esta capa también asegura que los datos cumplan con las normativas y políticas de privacidad.

Herramientas de seguridad:

• Apache Ranger y Kerberos son herramientas utilizadas para la autenticación y autorización de usuarios.

1.3 El Flujo de Datos en una Arquitectura de Big Data: Cómo se Mueven los Datos

El flujo de datos en una arquitectura de Big Data sigue una secuencia lógica:

1. Recopilación: Los datos se obtienen desde diversas fuentes.
2. Almacenamiento: Los datos se guardan en sistemas distribuidos como HDFS o bases de datos NoSQL.
3. Procesamiento: Los datos se procesan y transforman utilizando herramientas como Spark o MapReduce.
4. Análisis: Los datos procesados se analizan para obtener insights.
5. Visualización: Los resultados se presentan de manera comprensible para la toma de decisiones.

Esta arquitectura no solo permite manejar grandes volúmenes de datos, sino que asegura que sean accesibles, seguros y útiles para las organizaciones que desean aprovechar el poder de Big Data.

2.1 ¿Qué es Hadoop?

Hadoop es un marco de trabajo de código abierto diseñado para almacenar y procesar grandes volúmenes de datos de manera distribuida. Nació para abordar los desafíos del procesamiento de Big Data, permitiendo que las empresas manejen enormes cantidades de datos sin necesidad de usar costosos sistemas centralizados.

Hadoop está basado en la idea de computación distribuida, lo que significa que divide el trabajo entre varios nodos (computadoras) para realizar tareas en paralelo, aumentando la velocidad y la eficiencia. Está compuesto por varios módulos que permiten no solo almacenar grandes cantidades de datos, sino también procesarlos de forma rápida y fiable.

2.2 Clústeres de Hadoop: Estructura y Funcionamiento

Un clúster de Hadoop es un conjunto de nodos interconectados que trabajan juntos para almacenar y procesar grandes volúmenes de datos de manera distribuida. La arquitectura de Hadoop está diseñada para ser escalable, tolerante a fallos y eficiente, y se basa en dos tipos de nodos principales: nodos master y nodos slave.

2.2.1 Interacción entre el Cliente y el Master Node

Cuando un cliente (un usuario o una aplicación) quiere almacenar o procesar datos en un clúster de Hadoop, interactúa primero con el NameNode, que es el nodo maestro del sistema de archivos HDFS. El cliente puede realizar algunas consultas o peticiones como:

• Solicitar la ubicación de los datos: El cliente consulta al NameNode para obtener información sobre la ubicación de los bloques de datos dentro del clúster.
• Escribir y leer datos: Si el cliente desea escribir datos, primero consulta al NameNode para saber en qué nodos esclavos (DataNodes) debe almacenar los bloques. Si desea leer datos, el NameNode le proporciona la ubicación de los bloques en los DataNodes correspondientes.

2.2.2 Interacción entre el Master Node y los Slave Nodes

Los nodos master en un clúster de Hadoop son los responsables de gestionar la infraestructura y los recursos del clúster. Los nodos slave (DataNodes) son donde se almacenan realmente los datos y se realizan los procesos de cómputo.

NameNode y DataNodes:

• El NameNode gestiona la estructura del sistema de archivos (HDFS) y la metadata de los archivos, como la ubicación de los bloques de datos en los nodos esclavos.
• Los DataNodes almacenan los bloques de datos y responden a las solicitudes de lectura/escritura de los clientes.

Cuando un cliente solicita leer o escribir un archivo, el NameNode determina en qué DataNodes se encuentran los bloques de datos, y le informa al cliente sobre las ubicaciones. En el caso de la escritura, el cliente se conecta directamente a los DataNodes, que se encargan de almacenar los datos en los bloques.

2.3 Componentes de Hadoop

Existen 4 componentes de Hadoop que se pueden resumir en los siguiente:

• HDFS almacena los datos distribuidos en el clúster.
• YARN asigna recursos y coordina el procesamiento de datos.
• MapReduce ejecuta las tareas de análisis y procesamiento sobre los datos almacenados en HDFS.
• Hadoop Common actúa como base para garantizar que todo funcione de manera integrada.

2.3.1. HDFS (Hadoop Distributed File System)

El corazón del almacenamiento en Hadoop es el HDFS, un sistema de archivos distribuido diseñado para gestionar grandes volúmenes de datos.

• Función: Almacena los datos dividiéndolos en bloques que se distribuyen en múltiples nodos dentro de un clúster.
• Características Clave:
• Tolerancia a fallos: Cada bloque se replica (normalmente 3 veces) para garantizar que los datos no se pierdan si un nodo falla.
• Eficiencia: Permite acceder a grandes conjuntos de datos de forma rápida y distribuida.
• Bloques: El tamaño de los claves suele ser de 128 MB pero puede ser configurable.

2.3.2. MapReduce

MapReduce es el modelo de programación de Hadoop para el procesamiento de datos. Permite dividir tareas grandes en subtareas más pequeñas que se ejecutan en paralelo en los nodos del clúster.

• ¿Cómo funciona?
• Fase Map: Procesa los datos y los organiza en pares clave-valor.
• Fase Reduce: Toma los resultados de la fase Map, los agrupa y genera un resultado final.
• Ventaja: Es eficiente y escalable, ideal para análisis de grandes volúmenes de datos.

MapReduce divide las tareas en pequeños fragmentos y los distribuye en todos los nodos que tengamos, esto para ejecutarlo en paralelo. El resultado de cada MAP se vuelve un input para el proceso de reduce para obtener finalmente un resultado. Es así como se logra un procesamiento distribuido. Cabe resaltar que el procesamiento se realiza en disco NO SE ALMACENA EN MEMORIA, por eso es más lento que SPARK.

2.3.3. YARN (Yet Another Resource Negotiator)

YARN actúa como el sistema operativo del clúster de Hadoop, gestionando los recursos y coordinando la ejecución de tareas.

• Funciones:
• Administra los recursos del clúster.
• Asigna tareas de procesamiento a los nodos disponibles.
• Componentes Clave:
• ResourceManager: Coordina los recursos globales del clúster.
• NodeManager: Gestiona los recursos locales en cada nodo y ejecuta las tareas asignadas.

2.3.4. Hadoop Common

Este módulo incluye las bibliotecas y utilidades necesarias para que los otros componentes de Hadoop funcionen correctamente.

• Función Proporciona las herramientas esenciales para la integración y la comunicación entre los módulos.
• Ejemplo: Soporte para el sistema de archivos y la serialización de datos.

2.4 Aplicaciones Comunes del Ecosistema Hadoop

2.4.1. Ingesta de Datos (Data Ingestion)

Herramientas utilizadas para importar datos desde diversas fuentes al ecosistema Hadoop.

• Sqoop, Flume, Kafka

2.4.2. Almacenamiento Distribuido (Distributed Storage)

Sistemas diseñados para almacenar grandes volúmenes de datos de manera distribuida.

• HDFS, HBase

2.4.3. Programación Distribuida (Distributed Programming)

Herramientas que facilitan el procesamiento y análisis de datos en paralelo a través de múltiples nodos.

• MapReduce, Spark, Tez

2.4.4. Análisis de Datos (Data Analysis)

Aplicaciones que permiten consultar, transformar y analizar los datos almacenados en Hadoop.

• Hive, Pig, Impala

2.4.5. Orquestación y Gestión (Workflow Management)

Herramientas utilizadas para coordinar flujos de trabajo y gestionar tareas en un clúster de Hadoop.

• Oozie, Zookeeper

2.4.6. Visualización e Inteligencia de Negocios (BI)

Herramientas que permiten crear dashboards y visualizar los datos de forma interactiva.

• Tableau, QlikView

2.4.7. Machine Learning (Aprendizaje Automático)

Bibliotecas diseñadas para desarrollar modelos predictivos y aplicar técnicas de aprendizaje automático.

• Mahout, MLlib

3.1 Data Pipeline de MySQL a HDFS usando Sqoop

En el mundo del Big Data, trasladar datos desde bases de datos relacionales como MySQL hacia un sistema distribuido como HDFS es una tarea común y esencial para aprovechar el poder de análisis de Hadoop. Una herramienta clave para este proceso es Apache Sqoop, que permite transferir datos de manera eficiente entre sistemas.

3.1.1 Pasos para construir el Data Pipeline

1. Definir los parámetros de conexión Configura los detalles básicos como el host, puerto, base de datos, usuario, contraseña y el directorio de destino en HDFS.
2. Preparar el entorno Asegúrate de que Sqoop y HDFS estén correctamente instalados y configurados. Verifica también que haya suficiente espacio en HDFS para los datos.
3. Seleccionar la tabla o consulta Decide si transferirás una tabla completa o un subconjunto de datos mediante una consulta SQL personalizada.
4. Ejecutar el comando de Sqoop Utiliza Sqoop para importar los datos desde MySQL hacia HDFS. Puedes personalizar el proceso definiendo columnas específicas, delimitadores de campos, o configurando particiones con múltiples mappers para mejorar el rendimiento.
5. Verificar los datos en HDFS Comprueba que los datos se hayan transferido correctamente revisando el directorio de destino en HDFS.
6. Automatización y monitoreo Implementa scripts para ejecutar este proceso automáticamente según sea necesario. También puedes usar herramientas de orquestación como Apache Airflow o Oozie para programar, monitorizar y gestionar el pipeline.
7. Manejo de errores Asegura la integridad de los datos manejando errores durante el proceso, implementando notificaciones o reintentos automáticos si es necesario.

4. GUÍA PRÁCTICA PARA DESARROLAR EL DATA PIPELINE ENTRE MySQL Y HADOOP.

4.1 SUMMARY

La idea de esta guía práctica es conseguir extraer los datos de una base de datos llamada retail_db con ayuda de SQOOP para migrarlo a Hadoop y distribuir su almacencamiento con ayuda de HDFS para que pueda ser preparado para un posterior análisis con Hive y SPARK. El proceso se llevara con ayuda de contenedores de Docker y dentro de un repositorio de GitHub.

4.2 PRELIMINARES

Para empezar a trabajar con Big Data desplegaremos un cluster que contenga Hadoop, Hive y Spark. Los archivos necesarios para trabajar en esto ambientes pueden ser forkeados en el repositorio.

[embed]GitHub - Adrian-Valensia/BigDataDeveloper Contribute to Adrian-Valensia/BigDataDeveloper development by creating an account on GitHub.github.com

4.2.1 Archivo Docker-compose en el repositorio

Para entablar comunicación entre Hadoop, Hive y Spark necesitamos descargar las imagenes de las mismas haciendo uso de los contenedores definidos en el archivo docker-compose.yml. Explorando el archivo es posible ver que contamos con:

1. • Definición de los nodos master y slave de nuestro cluster:

2.- Parámetros de Conexión con MySQL, para conectarnos a la base de datos de donde extraeremos los datos.

3.- Creación de la imágen de Hive-server: permite a las aplicaciones cliente (como JDBC o ODBC) interactuar con Hive a través de una interfaz de consultas. Su función principal es manejar las solicitudes de ejecución de consultas y la conexión de los clientes a Hive.

4.- Contenedor para crear la imágen de la metadata de hive.

5.- El contenedor hive-metastore-postgresql proporciona un entorno para almacenar los metadatos de Hive (información sobre las tablas, particiones, etc.) utilizando PostgreSQL como su base de datos de almacenamiento.

6.- El contenedor de jupyter, que se utiliza comunmente para análisis de datos.

7.- Spark-master y spark-worker son nodos amo y esclavo, respectivamente, de Spark.

El resultado son distintos contenedores con distintos servicios simulando un cluster de Big Data.

4.2.2 Extensiones necesarias

Al estar trabajando en GitHub es de mucha utilidad descargar dos extensiones:

1.- MySQL Database Management for MySQL/MariaDB, PostgreSQL, Redis and ElasticSearch.

2.- Docker Explorer: Administra imágenes y contenedores de Docker.

4.3 Integración de nuestro DataPipeline

4.3.1 Inicializar el archivo Docker compose

para ello necesitamos escribir en la terminal el comando:

docker compose up

después de cierto tiempo los contenedores estarán habilitados y eso se puede verificar con ayuda del doker explorer en la esquina inferior izquierda del codespace.

4.3.2 Explorar las carpetas del repositorio

Es necesario dar un vistazo primero a las carpetas o archivos que se encuentran de manera local en nuestro repositorio. Esto se hace para mantener a la vista el contenido local respecto al que se encuentra en nuestro ambiente hadoop. Para ello abriremos una nueva terminal y escribiremos el comando

ls

esto desplegara el siguiente contenido:

es posible darnos cuenta que aparece los archicvos almacenados en la carpeta BigDataDeveloper como nuestro archivo README.md y algunas carpetas. La carpeta mysql es donde se almacenara nuestra base de datos.

4.3.3 Habilitar el ambiente de Hadoop

Una vez que notamos la ubicación de las carpetas y documentos de nuestro repositorio, tenemos que abrir una terminal nueva y ejecutar el siguiente comando para habilitar el ambiente Hadoop.

docker exec -it datanode bash

El output debe ser similar a este:

Cabe resaltar que al encontrarnos en el ambiente de Hadoop, la manera de acceder a las carpetas y archivos cambiará. Para listar los archivos y carpetas, en este caso, se utiliza el comando:

hdfs dfs -ls /

Al aplicarlo se vería algo como lo siguiente.

lo cual nos indica que solo tenemos la carpeta user y la de temporales.

4.3.4 Comandos útiles en HDFS

Dentro de HDFS los comando son muy parecidos a los de Linux, para los fines de este blog bastaría con aprenderse el comando de lectura ‘ls’ y los de visualización ‘cat’ y ‘text’.

Para mostrar el contenido del directorio donde te encuentras:

hdfs dfs -ls /path

Para leer el contenido de un archivo:

hdfs dfs -cat /path/file

para ver el contenido de un archivo en formato de texto:

hdfs dfs -text /path/file

para crear un directorio en una ruta especifica

hdfs dfs -mkdir /path/nombre_del_directorio_nuevo

El método PUT se utiliza para cargar archivos desde el sistema de archivos local al sistema de archivos distribuido HDFS de Hadoop.

hdfs dfs -put <archivo_local> <ruta_destino_hdfs>

El método GET se utiliza para descargar archivos o directorios desde el sistema de archivos distribuido HDFS hacia el sistema de archivos local.

hdfs dfs -get <ruta_origen_hdfs> <ruta_destino_local>

Otra forma de dubir archivos de HDFS a local y viceversa es cón los métodos -copyFromLocal y -copyToLocal.

4.3.5 Habilitar SQOOP para la ingesta de datos

Para habilitar la herramienta de SQOOP dentro del ambiente de Hadoop tenemos que aplicar el comando:

sh /datanode/scripts/scripts.sh

El output se vería así:

4.3.6 Crear y ejecutar los scripts que conectan con la base de datos.

En nuestro caso, ya tenemos creados los scripts guardados en la carpeta con el mismo nombre, que corresponden a los archivos script_sqoop_avro.sh y script_sqoop_textfile.sh.

El contenido de ambos scripts es muy similar. Por ejemplo, en script_sqoop_textfile.sh podemos encontrar este pedazo de código.

La IP en este caso corresponde a 10.0.1.8

que lo que nos dice es que nos conectemos a una base de datos con ciertos parámetros y que extraiga el contenido de la tabla customers en el PATH de Hadoop donde queremos que se encuentre. Este proceso se repite para todas las tablas de la base de datos que podemos consultar haciendo uso de la extensión de mysql.

En este paso es importante aclarar que cada nueva conexión que se haga dentro de GitHub nos creará una IP nueva, por lo que tenemos que actualizarla y recolocarla en el script. En nuestro caso, podemos encontrarla con el comando en una nueva terminal bash:

ifconfig

La IP actualizada saldrá en el apartado eth0 como inet:

Ahora, podemos usar esta misma IP para acceder a nuestra extensión de mysql y verificar qué tablas contiene nuestra base de datos (en nuestro caso retail_db):

tablas de nuestra base de datos retail_db

Ahora, ya que modificamos las IP de nuestro script de todas las tablas que exportaremos como archivo de texto haremos lo mismo para las tablas que exportaremos como tipo avro que se encuentran en el script script_sqoop_avro.sh.

Es posible observar que el script de avro hace exactamente lo mismo.

se conecta a la base de datos y extrae la tabla customers en formato avro a la ruta de Hadoop especificada. Tener presente la correción de IP cada que se abra GitHub.

Un método para verificar que ya estamos conectados a MySQL en Hadoop es aplicar el comando:

sqoop list-tables \
--connect "jdbc:mysql://10.0.1.8:3310/retail_db" \
--username root \
--password root

El output serán las tablas de nuestra base de datos:

Ahora,para ejecutar los dos scripts, debemos aplicar dentro del ambiente Hadoop (no en bash) los comandos:

sh /datanode/scripts/sqoop/script_sqoop_textfile.sh
sh /datanode/scripts/sqoop/script_sqoop_avro.sh

Cuando se aplica el comando sh para cargar el script de avro, es posible ver que durante la ejecución se crean y eliminan archivos, muchos de ellos .java. SQOOP es la herramienta que genera los scripts finales, cosa que antes se tenía que hacer manualmente.

4.3.7 Verificar la extracción en Hadoop (HDFS)

Recordemos que en los scripts dirigimos la extracción a un PATH determinado dentro de HADOOP.

Textfile -> /user/datapath/datasets/

Avro -> /user/datapath/datasets/avro

Entonces usaremos los comando “tipo Linux” dentro de Hadoop para hacer las consultas.

hdfs dfs -ls /user/datapath/datasets/
hdfs dfs -ls /user/datapath/datasets/avro

En el primer caso obtendriamos el siguiente output:

y si quisieramos ir más a fondo consultando el contenido de una tabla (como customers) utilizariamos el comando:

hdfs dfs -ls /user/datapath/datasets/customers

Lo que muestra es nuestro archivo dividido en 4 partes, para tener acceso a cada una, podríamos usar el comando cat.

hdfs dfs -cat hdfs dfs -ls /user/datapath/datasets/customers/part-m-00000

Ahora podríamos repetir el proceso para los archivos avro.

1. Accedemos a la ubicación del script

hdfs dfs -ls /user/datapath/datasets/avro

2.- Indagamos en cualquiera de las tablas, p. ej. Customers

hdfs dfs -ls /user/datapath/datasets/avro/customers

3.- Elegimos cualquiera de los bloques en los que se ha dividido el archivo y miramos el contenido con “cat” o “text”.

hdfs dfs -text /user/datapath/datasets/avro/customers/part-m-00000.avro

Como text interpreta archivo de texto, por eso le cambia el formato a la data.

Así, con esta ingesta de datos ya podemos utilizarlos para analizarlo con Hive o Spark.

Conclusión

En este artículo se crearon pipelines de datos desde MySQL hacia HDFS, se detallaron los pasos para transferir datos de una base de datos MySQL a un sistema de archivos distribuido HDFS, facilitando el análisis de grandes volúmenes de información. Este proceso comenzó con la instalación y configuración de Sqoop, una herramienta diseñada para la transferencia eficiente de datos entre bases de datos relacionales y Hadoop. Además, se estableció la conexión entre MySQL y HDFS, especificando las credenciales y parámetros necesarios. Posteriormente, se ejecutó el comando sqoop import para extraer los datos de las tablas seleccionadas en MySQL y cargarlos en HDFS, organizándolos en directorios y archivos adecuados para su procesamiento posterior. Este enfoque permitió integrar datos estructurados de MySQL en el ecosistema Hadoop, dando pauta a análisis de Big Data mediante Hive o Spark.