Introducción al uso de herramientas para

Big Data de la tecnologı́a Hadoop

Salomón Ramı́rez*
Seminario: Spatial Big Data
Especialización en Sistemas de Información Geográfica
Facultad de Ingenierı́a
Universidad Distrital Francisco José de Caldas
Bogotá, Colombia
Octubre de 2019

Contenido distribuido bajo Licencia Creative Commons

C
CC BY:

Resumen
Este documento brinda una breve introducción al uso de las herra-
mientas de código abierto más populares que hacen parte del ecosistema
de Hadoop en el contexto de Big Data, las cuales están incluidas en el
ambiente de Cloudera Distribution Hadoop (CDH).
Especı́ficamente se realizan algunos casos de uso analı́tico tomados y
adaptados de la documentación oficial de Cloudera con fines académicos,
sin pretender ser una guı́a exhaustiva, ni remplazar la documentación
oficial.
Se asume que el lector esta familiarizado con los fundamentos de Big
Data, la arquitectura de Haddop, el uso de MapReduce y tiene conceptos
y habilidades tanto en programación como en bases de datos relacionales.

Introducción
En cualquier prueba de concepto (PoC ) exitosa se deben abordar problemáti-
cas que sean de interés para la organización. Por lo tanto, lo primero que debe
hacer es definir una pregunta de negocio (BQ).
Los ejercicios se realizaran en el contexto de una empresa llamada DataCo,
y su misión como Especialista es ayudar a esta empresa a obtener una mejor
visión, formulando y resolviendo grandes interrogantes de negocio.

* [email protected]

1
 Figura 1: Modelo de datos relacional de DataCo. Fuente: Cloudera, 2017

1. Análisis de datos estructurados

BQ: ¿Cuáles son los productos más vendidos?

1. Describa cómo resolverı́a el interrogante planteado anteriormente, bajo un

enfoque tradicional de gestión y procesamiento de datos.
2. Abra una terminal o Interfaz de Linea de Comandos (CLI).

3. Realice ingesta de datos estructurados en Hive a partir de datos almacena-

dos en un Sistema Manejador de Bases de Datos Relacionales (SMBDR)
empleando el comando import-all-tables de Sqoop mediante las siguientes
instrucciones:

sqoop import-all-tables \
-m 1 \
--connect jdbc:mysql://10.0.2.15:3306/retail_db \
--username=retail_dba \
--password=cloudera \
--compression-codec=snappy \
--as-parquetfile \
--warehouse-dir=/user/hive/warehouse \
--hive-import

4. Describa cada uno de los parámetros empleados en la instrucción anterior.

5. ¿Sobre cuál herramienta de Hadoop se realizó la ingesta?

6. Confirme que los datos fueron importados en HDFS mediante las siguien-
tes instrucciones, una vez finalice la ejecución de las anteriores:

2
 hadoop fs -ls /user/hive/warehouse/
hadoop fs -ls /user/hive/warehouse/categories/
7. Describa el resultado obtenido en cada caso.
8. Abra un navegador web e ingrese a http://10.0.2.15:8888 para iniciar Hue

9. Autentifı́quese usando tanto para usuario como para constraseña cloudera

como credenciales.
10. Abra el editor Impala que se encuentra en el menú Query Editors de Hue
11. Actualice los metadatos de Impala ejecutando la siguiente instrucción:

invalidate metadata;

12. Describa el resultado obtenido anteriormente.

13. Liste las tablas existentes empleando la siguiente instrucción

show tables;
14. Recupere los 10 productos más vendidos mediante la siguiente consulta:

-- categorias productos más populares

select c.category_name, count(order_item_quantity) as count
from order_items oi
inner join products p on oi.order_item_product_id = p.product_id
inner join categories c on c.category_id = p.product_category_id
group by c.category_name
order by count desc
limit 10;

15. Recupere los 10 productos que generan mayores ingresos, empleando la

siguiente consulta:

-- Principales productos generadores de ingresos

select p.product_id, p.product_name, r.revenue
from products p inner join
(select oi.order_item_product_id,
sum(cast(oi.order_item_subtotal as float)) as revenue
from order_items oi inner join orders o
on oi.order_item_order_id = o.order_id
where o.order_status <> ’CANCELED’
and o.order_status <> ’SUSPECTED_FRAUD’
group by order_item_product_id) r
on p.product_id = r.order_item_product_id
order by r.revenue desc
limit 10;

3
 16. Plante y resuelva otro BQ para DataCo que sea de su interés como Espe-
cialista, considerando para ello el modelo de datos relacional de DataCo
y usando SQL.
17. ¿Por qué si la ingesta de datos se realizó sobre Hive es posible consultarlos
desde Impala ?
18. Formule conclusiones.

2. Análisis de datos no estructurados

BQ: ¿Los productos más vendidos son también los productos más buscados
o vistos?

Para determinar qué es lo que más ven o buscan los visitantes del sitio web
de DataCo, se pueden emplear los datos de actividad web (clickstream).

1. Abra la una CLI.

2. Verifique que el archivo de log se encuentre en la ruta /opt/examples/log data/
ejecutando la siguiente instrucción:
ls /opt/examples/log_files/
3. Recupere el primer registro almacenado ejecutando la siguiente instruc-
ción:
head -1 /opt/examples/log_files/access.log.2
4. Recupere el ultimo registro almacenado ejecutando la siguiente instruc-
ción:
tail -1 /opt/examples/log_files/access.log.2
5. Indique cuál es el periodo de análisis, a partir de la fecha y hora de cada
registro recuperado anteriormente.
6. Describa el contenido del archivo log e indique el tipo de estructura de
datos que posee.
7. Cree un directorio en le sistemas de archivos de Hadoop ejecutando ls
siguiente instrucción:
sudo -u hdfs hadoop fs -mkdir /user/hive/warehouse/original_access_logs
8. Copie los datos desde sistema del archivos local hacia el sistema de archivos
de Hadoop, ejecutando la siguiente instrucción:
sudo -u hdfs hadoop fs -copyFromLocal \
/opt/examples/log_files/access.log.2 \
/user/hive/warehouse/original_access_logs

4
 9. Compruebe que los datos están en sistema de archivos de Hadoop ejecu-
tando la siguiente instrucción:
hadoop fs -ls /user/hive/warehouse/original_access_logs
10. Abra el editor consultas Hive que se encuentra en el menú Query Editors
de Hue y ejecute las siguientes instrucciones, para crear una tabla en Hi-
ve que permita consultar estos datos a través de Impala. La creación de
la tabla se realizará en dos pasos. Inicialmente, se aplicaran SerDes (se-
rializadores/deserializadores) flexibles de Hive para tokenizar los registros
de acceso en campos individuales en una tabla intermedia (intermedia-
te access logs), a partir del uso de expresiones regulares. Luego, se ingesta-
ran los datos desde la tabla intermedia a una tabla (tokenized access logs)
que no requiere ningún SerDe especial.

--Tabla intermedia SerDes

CREATE EXTERNAL TABLE intermediate_access_logs (
ip STRING,
date STRING,
method STRING,
url STRING,
http_version STRING,
code1 STRING,
code2 STRING,
dash STRING,
user_agent STRING)
ROW FORMAT SERDE ’org.apache.hadoop.hive.contrib.serde2.RegexSerDe’
WITH SERDEPROPERTIES (
’input.regex’
=
’([^ ]*) - - \\[([^\\]]*)\\] "([^\ ]*) ([^\ ]*) ([^\ ]*)"
(\\d*) (\\d*) "([^"]*)" "([^"]*)"’,
’output.format.string’ = "%1$$s %2$$s %3$$s %4$$s %5$$s %6$$s %7$$s %8$$s %9$$s")
LOCATION ’/user/hive/warehouse/original_access_logs’;

--Tabla final tokenizada

CREATE EXTERNAL TABLE tokenized_access_logs (
ip STRING,
date STRING,
method STRING,
url STRING,
http_version STRING,
code1 STRING,
code2 STRING,
dash STRING,
user_agent STRING)

5
 ROW FORMAT DELIMITED FIELDS TERMINATED BY ’,’
LOCATION ’/user/hive/warehouse/tokenized_access_logs’;

--Importación librerı́as
ADD JAR /usr/lib/hive/lib/hive-contrib.jar;

--Inserción datos
INSERT OVERWRITE TABLE tokenized_access_logs
SELECT * FROM intermediate_access_logs;

11. Ingrese al editor de consultas Impala, actualice los metadatos y liste las
tablas.

12. Describa el resultado obtenido anteriormente.

13. ¿Cuántos registros tiene la tabla tokenized access logs?.
14. Explore el contenido de la tabla tokenized access logs y describa su estruc-
tura.

15. Recupere los productos más buscados o vistos, mediante la siguiente con-
sulta:
select count(*),url from tokenized_access_logs
where url like ’%\/product\/%’
group by url order by count(*) desc;

16. Genere una consulta de su interés como Especialista, que permita resolver
otro BQ de DataCo, a partir del contenido de los registros de acceso.
17. Realice un análisis introspectivo de los resultados obtenidos que describa
la relación existente entre los productos más vendidos y los productos más
buscados o vistos.

18. ¿Existe algún producto de los más vistos que no se encuentre entre los
más vendidos?
19. Formule conclusiones.

3. Analı́tica avanzada: Análisis de correlación

BQ: ¿Cuales son las parejas de productos comprados juntos con mayor fre-
cuencia?

1. Inicie una terminal de Spark desde una CLI empleado la siguiente instruc-
ción:

6
 spark-shell --master yarn-client
2. Ejecute las siguientes sentencias de código del lenguaje de programación
Scala:

//importación librerı́a
import org.apache.hadoop.mapreduce.Job
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat
import org.apache.avro.generic.GenericRecord
import parquet.hadoop.ParquetInputFormat
import parquet.avro.AvroReadSupport
import org.apache.spark.rdd.RDD

//Creación RDD’s tablas order_items/products

def rddFromParquetHdfsFile(path: String): RDD[GenericRecord] = {
val job = new Job()
FileInputFormat.setInputPaths(job, path)
ParquetInputFormat.setReadSupportClass(job,
classOf[AvroReadSupport[GenericRecord]])
return sc.newAPIHadoopRDD(job.getConfiguration,
classOf[ParquetInputFormat[GenericRecord]],
classOf[Void],
classOf[GenericRecord]).map(x => x._2)
}

val warehouse = "hdfs://quickstart/user/hive/warehouse/"

val order_items = rddFromParquetHdfsFile(warehouse + "order_items");
val products = rddFromParquetHdfsFile(warehouse + "products");

//Extracción campos tablas order_items/products

val orders = order_items.map { x => (
x.get("order_item_product_id"),
(x.get("order_item_order_id"), x.get("order_item_quantity")))
}.join(
products.map { x => (
x.get("product_id"),
(x.get("product_name")))
}
).map(x => (
scala.Int.unbox(x._2._1._1), // order_id
(
scala.Int.unbox(x._2._1._2), // quantity
x._2._2.toString // product_name
)

7
 )).groupByKey()

//Conteo número veces aparece combicación productos juntos (10 más frecuentes)
val cooccurrences = orders.map(order =>
(
order._1,
order._2.toList.combinations(2).map(order_pair =>
(
if (order_pair(0)._2 < order_pair(1)._2)
(order_pair(0)._2, order_pair(1)._2)
else
(order_pair(1)._2, order_pair(0)._2),
order_pair(0)._1 * order_pair(1)._1
)
)
)
)
val combos = cooccurrences.flatMap(x => x._2).reduceByKey((a, b) => a + b)
val mostCommon = combos.map(x => (x._2, x._1)).sortByKey(false).take(10)

// Impresión resultados
println(mostCommon.deep.mkString("\n"))

3. Salga del shell de Spark usando el siguiente comando:

exit
4. Analice y describa los resultados obtenidos anteriormente.
5. Formule conclusiones.

4. Recolección y análisis de datos en tiempo real

1. Cargue la configuración de Solr, mediante las siguientes instrucciones:
cd /opt/examples/flume
solrctl --zk quickstart:2181/solr instancedir --create live_logs ./solr_configs
2. Cree una colección en Solr, usando la siguiente instrucción:
solrctl --zk quickstart:2181/solr collection --create live_logs -s 1
3. Compruebe que ha creado correctamente la colección en Solr, accediendo
a Hue y haciendo clic en la opción Search del menú superior.
4. Realice clic en el botón Index que se encuentra en la parte superior dere-
cha, para ver todos los ı́ndices/colecciones.

8
 5. Realice clic sobre el indice live logs creado anteriormente.
6. Inicie el generador de registro de actividad desde una CLI, con el siguiente
comando:
start_logs

7. Verifique el funcionamiento del generador de registros, usando el siguiente

comando:
tail_logs
8. Describa los resultados obtenidos anteriormente.

9. Inicie el agente de recolección de datos en tiempo real Flume, usando las

siguientes instrucciones:
flume-ng agent \
--conf /opt/examples/flume/conf \
--conf-file /opt/examples/flume/conf/flume.conf \
--name agent1 \
-Dflume.root.logger=DEBUG,INFO,console
10. Describa los resultados obtenidos anteriormente.
11. Realice clic en la opción Search que se encuentra debajo de Actions en
el panel izquierdo de Hue, para buscar y explorar los eventos que se han
indexado.
12. Habilite el modo de edición realizando clic sobre el icono de lápiz (Edit)
que se encuentra en la parte superior derecha.
13. Arrastre y suelte un gráfico de tipo Bar Chart a la caja gris horizontal,
que aparece debajo de los tipos de gráficos dispuestos en la parte superior.
14. Seleccione como campo de agrupación, el campo request date de las op-
ciones mostradas en la parte superior.
15. Describa los resultados obtenidos anteriormente.

16. Seleccione el diseño (Layout) de dos columnas, en la parte superior iz-

quierda.
17. Arrastre y suelte un gráfico de tipo Pie Chart a la fila recién creada en el
panel izquierdo.
18. Seleccione como campo de agrupación, el campo departament de las op-
ciones mostradas en la parte superior.
19. Describa los resultados obtenidos anteriormente.
20. Añada un filtro de tipo Text Facet en la segunda fila del panel izquierdo.

9
21. Seleccione como faceta el campo product, de las opciones mostradas en la
parte superior.
22. Describa los resultados obtenidos anteriormente.
23. Realice clic nuevamente en el icono de lápiz, para salir del modo de edición.

24. Construya un nuevo tablero con otros gráficos y campos de su interés.

25. Formule conclusiones.

10