Monta una Infraestructura para Big Data en tu Empresa

Taller:

“Monta una infraestructura Big Data para tu Empresa”

Urko Zurutuza

Dpto. Electrónica e Informática

Mondragon Goi Eskola Politeknikoa JMA

Mondragon Unibertsitatea

2

Enlaces

• http://bit.ly/big_data_sesion3

• http://bit.ly/ejercicios_bigdata

http://bit.ly/big_data_sesion3

http://bit.ly/ejercicios_bigdata

Agenda

• Día 1: Introducción al Big Data

• Día 2: Instalación de un cluster Cloudera

• Día 3: Ejercicios sobre Cloudera

Índice

1. Ejemplo de MapReduce

2. Creando y editando una base de datos HBase

3. Ingerir datos estructurados con Apache sqoop

4. Consultar datos estructurados con HIVE e Impala

5. El valor del Big Data: datos no estructurados

6. Análisis avanzado con Spark

7. Búsquedas en tiempo real con Solr

8. Creación de un panel de mandos

Consideraciones previas

• Cloudera Quick Start: • http://www.cloudera.com/content/cloudera/en/documen

tation/core/latest/topics/cloudera_quickstart_vm.html

• Dirección Cloudera Manager: http://ip_cloudera:7180

• Dirección Hue: http://ip_cloudera:8888

• Algunos links no son relativos (y erroneamentellevarán a quickstart.cloudera)

http://www.cloudera.com/content/cloudera/en/documentation/core/latest/topics/cloudera_quickstart_vm.html

http://193.146.78.55:7180

http://193.146.78.55:8888

http://193.146.78.55:7180

http://193.146.78.55:8888

Consideraciones previas

• Cuenta (con privilegios sudo):• username: cloudera

• password: cloudera

• La contraseña de root de MySQL es la misma, al igual que Hue y ClouderaManager

• Consideraciones previas

Ejercicio 1: MapReduce

Procesamiento en batch: MapReduce

• Marco para escribir aplicaciones que procesanenormes cantidades de datos de forma paralelaen grandes clusters formados por hardware común.

• MapReduce job• Divide el conjunto de datos de entrada en pedazos

(chunks) que son procesados por tareas “map” de forma paralela.

• Ordena las salidas de los “maps”, que son entrada a las tareas “reduce”.

• Opera sobre pares clave-valor.• No necesariamente en Java.

Procesamiento en batch: MapReduce

, 4

, 2

, 4

, 3

, 4

, 2

, 4

, 3

Split MapShuffle

&

Sort

Reduce

Master

assign

sassigns

Procesamiento en batch: MapReduce

• Ejemplo:

$ bin/hadoop dfs -cat /usr/joe/wordcount/input/file01

Hello World Bye World

$ bin/hadoop dfs -cat /usr/joe/wordcount/input/file02

Hello Hadoop Goodbye Hadoop

Procesamiento en batch: MapReduce

• Mapper: • procesa cada línea (cada línea un map)• Divide en tokens separados por espacios (palabras)• Emite un conjunto de claves-valor, en modo <palabra, 1>

• Map 1:• < Hello, 1>

• < World, 1>

• < Bye, 1>

• < World, 1>

• Map 2:• < Hello, 1>

• < Hadoop, 1>

• < Goodbye, 1>

• < Hadoop, 1>

Procesamiento en batch: MapReduce

• Combiner:• Cada map se ordena y genera agregados locales:

• Salida del primer Map:• < Bye, 1>

• < Hello, 1>

• < World, 2>

• Salida del segundo Map:• < Goodbye, 1>

• < Hadoop, 2>

• < Hello, 1>

Procesamiento en batch: MapReduce

• Reducer:• Suma los valores (ocurrencias de cada clave):

• < Bye, 1>

• < Goodbye, 1>

• < Hadoop, 2>

• < Hello, 2>

• < World, 2>

Ejercicio

• WordCount

• Documento ejercicios.doc

18

Ejercicio 1: Archivos de entrada

19

Ejercicio 1: ejecuciónMapReduce

[cloudera@quickstart ~]$ hadoop jar wordcount.jar org.myorg.WordCount /user/cloudera/wordcount/input

/user/cloudera/wordcount/output

16/06/29 10:21:52 INFO client.RMProxy: Connecting to ResourceManager at /0.0.0.0:8032

16/06/29 10:21:52 WARN mapreduce.JobResourceUploader: Hadoop command-line option parsing not performed. Implement the Tool

interface and execute your application with ToolRunner to remedy this.

16/06/29 10:21:52 INFO mapred.FileInputFormat: Total input paths to process : 2

16/06/29 10:21:53 INFO mapreduce.JobSubmitter: number of splits:3

16/06/29 10:21:53 INFO mapreduce.JobSubmitter: Submitting tokens for job: job_1467220526122_0001

16/06/29 10:21:53 INFO impl.YarnClientImpl: Submitted application application_1467220526122_0001

16/06/29 10:21:53 INFO mapreduce.Job: The url to track the job:

http://quickstart.cloudera:8088/proxy/application_1467220526122_0001/

16/06/29 10:21:53 INFO mapreduce.Job: Running job: job_1467220526122_0001

16/06/29 10:22:03 INFO mapreduce.Job: Job job_1467220526122_0001 running in uber mode : false

16/06/29 10:22:03 INFO mapreduce.Job: map 0% reduce 0%

16/06/29 10:22:17 INFO mapreduce.Job: map 33% reduce 0%

16/06/29 10:22:19 INFO mapreduce.Job: map 67% reduce 0%

16/06/29 10:22:20 INFO mapreduce.Job: map 100% reduce 0%

16/06/29 10:22:25 INFO mapreduce.Job: map 100% reduce 100%

16/06/29 10:22:25 INFO mapreduce.Job: Job job_1467220526122_0001 completed successfully

16/06/29 10:22:25 INFO mapreduce.Job: Counters: 49

File System Counters

FILE: Number of bytes read=79

FILE: Number of bytes written=455843

FILE: Number of read operations=0

FILE: Number of large read operations=0

FILE: Number of write operations=0

HDFS: Number of bytes read=410

HDFS: Number of bytes written=41

HDFS: Number of read operations=12

HDFS: Number of large read operations=0

HDFS: Number of write operations=2

20

Ejercicio 1: ejecuciónMapReduceJob Counters

Launched map tasks=3

Launched reduce tasks=1

Data-local map tasks=3

Total time spent by all maps in occupied slots (ms)=36685

Total time spent by all reduces in occupied slots (ms)=5500

Total time spent by all map tasks (ms)=36685

Total time spent by all reduce tasks (ms)=5500

Total vcore-seconds taken by all map tasks=36685

Total vcore-seconds taken by all reduce tasks=5500

Total megabyte-seconds taken by all map tasks=37565440

Total megabyte-seconds taken by all reduce tasks=5632000

Map-Reduce Framework

Map input records=2

Map output records=8

Map output bytes=82

Map output materialized bytes=91

Input split bytes=357

Combine input records=8

Combine output records=6

Reduce input groups=5

Reduce shuffle bytes=91

Reduce input records=6

Reduce output records=5

Spilled Records=12

Shuffled Maps =3

Failed Shuffles=0

Merged Map outputs=3

GC time elapsed (ms)=445

CPU time spent (ms)=1970

Physical memory (bytes) snapshot=762191872

Virtual memory (bytes) snapshot=6006784000

Total committed heap usage (bytes)=557592576

21

Ejercicio 1: ejecuciónMapReduce

Shuffle Errors

BAD_ID=0

CONNECTION=0

IO_ERROR=0

WRONG_LENGTH=0

WRONG_MAP=0

WRONG_REDUCE=0

File Input Format Counters

Bytes Read=53

File Output Format Counters

Bytes Written=41

22

Ejercicio 1: Archivos de salida

23

Ejercicio 1: resultado

[cloudera@quickstart ~]$ hadoop fs -cat /user/cloudera/wordcount/output/part-

00000

Bye 1

Goodbye 1

Hadoop 2

Hello 2

World 2

Ejemplo HBase

HBase

• Qué es Hbase?• En Hbase los datos se almacenan en tablas, las cuales

tienen filas y columnas… pero mejor verlo como un mapa u objeto multidimensional

• Una tabla de Hbase contiene multiples filas.

• Una fila de datos tiene una clave ordenable y un número arbitrario de columnas. La tabla se almacenade forma dispersa, de modo que las filas de unamisma tabla pueden tener columnas tan variadascomo se quiera.

HBase

• Map: filas de datos

{

"zzzzz" : "woot",

"xyz" : "hello",

"aaaab" : "world",

"1" : "x",

"aaaaa" : "y"

}

HBase

• Sorted Map: filas ordenadas por la clave

{

"1" : "x",

"aaaaa" : "y",

"aaaab" : "world",

"xyz" : "hello",

"zzzzz" : "woot"

}

HBase

• Multidimensional: familias de columnas por cadafila{

"1" : {

"A" : "x",

"B" : "z"

},

"aaaaa" : {

"A" : "y",

"B" : "w"

},

"aaaab" : {

"A" : "world",

"B" : "ocean"

},

"xyz" : {

"A" : "hello",

"B" : "there"

},

"zzzzz" : {

"A" : "woot",

"B" : "1337"

}

HBase

• Multidimensional: una familia de columnas puedetener varias columnas, identificadas por unaetiqueta o qualifier{

// ...

"aaaaa" : {

"A" : {

"foo" : "y",

"bar" : "d"

},

"B" : {

"" : "w"

}

},

"aaaab" : {

"A" : {

"foo" : "world",

"bar" : "domination"

},

"B" : {

"" : "ocean"

}

},

// ...

}

HBase

• Multidimensional: timestamp{

// ...

"aaaaa" : {

"A" : {

"foo" : {

15 : "y",

4 : "m"

},

"bar" : {

15 : "d",

}

},

"B" : {

"" : {

6 : "w"

3 : "o"

1 : "w"

}

},

// ...

}

Ejemplo con HBase

Ejemplo con HBaseConectarse a HUE: http://localhost:8888

32

http://localhost:8888/

Ejemplo con HBase

33

Ejemplo con HBase

34

Ejemplo con HBase

35

Ejemplo con HBase

36

Ejemplo con HBase

37

Ejemplo con HBase

38

Ejemplo con HBase

39

Ejemplo con HBase

40

Ejemplo con HBase

41

Ejemplo con HBase

42

Ejercicio 2: Ingerir datos estructurados

Un paseo por la potencia de Big Data

• Imaginemos que somos el “Data Specialist” de una corporación

• El análisis de los datos debe responder a preguntas concretas

Definir una pregunta de Negocio

• Sin una pregunta bien definida:• No sabremos como modelar los datos

• Qué estructuras de datos aplicar

• Qué conjuntos de datos deberemosobtener

• Qué herramientas necesitaremos

Ingerir datos estructurados

• Pregunta de Negocio: “Qué productos les gustacomprar a nuestros clientes?”

• Esta información se guarda por lo general en unabase de datos Transaccional, la cual nos dirá quéclientes compran y quienes no cada producto.

• Necesitamos ingerir los datos de una BBDD en HDFS

sqoop

Ingerir datos estructurados

• Sqoop nos permite cargar los datos de MySQL en HDFS, preservando la estructura de las tablas.

• Con unos parámetros extra, podremos cargarlosen un formato (avro) que podrá luego ser utilizadopor Impala

Ingerir datos estructurados: sqoop

tablas tablasFicheros

distribuidos

BBDD Relacional

BBDD RelacionalHDFS

Tratamiento

(Map-Reduce) (Map-Reduce)

(Map-Reduce)

Ejercicio 2: Ingerir datos estructurados: sqoop

• Pasos:1. Importar todas las tablas desde MySQL a

HDFS.

2. Sqoop generará los esquemas de cada tabla.Copiar estos esquemas a HDFS para queHIVE los tenga disponibles.

3. Arrancamos Hive y definimos las tablas. Dosopciones: (1) desde la línea de comandos y(2) desde la interfaz web.

4. Mostramos las tablas

50

Ingerir datos estructurados

(Mirar Documento)

[root@quickstart.cloudera ~] sqoop import-all-tables \

-- num-mappers 1 \

--connect jdbc:mysql://quickstart.cloudera:3306/retail_db \

--username=retail_dba \

--password=cloudera \

--compression-codec=snappy \

--as-avrodatafile \

--warehouse-dir=/user/hive/warehouse/userXX

Ejercicio 2: Ingerir datos estructurados: sqoop

52

Ejercicio 2: Ingerir datos estructurados: sqoop

53

Ejercicio 2: Definir las tablas: Hive

54

Ejercicio 2: Definir las tablas: Hive

55

Ejercicio 2: Definir las tablas: Hive

56

Ejercicio 2: Mostrar las tablas: Hive

57

Ingerir datos estructurados

Ejercicio 3: Consultar datos estructurados

Consultar datos estructurados

• Dos herramientas para hacer consultas SQL en CDH:

• Hive • Traduce las consultas SQL a trabajos MapReduce. • Bueno para trabajos en lote grandes, aplicando

transformaciones de forma flexible.

• Impala• Significativamente más rápido y preparado para tener

baja latencia para consultas interectivas y exploraciónde datos.

• Ambos comparten los metadatos sobre las tablas

Consultar datos estructurados

• Ejemplo en HIVE:[cloudera@quickstart ~]$ hive

CREATE EXTERNAL TABLE categories

ROW FORMAT SERDE 'org.apache.hadoop.hive.serde2.avro.AvroSerDe'

STORED AS INPUTFORMAT 'org.apache.hadoop.hive.ql.io.avro.AvroContainerInputFormat'

OUTPUTFORMAT 'org.apache.hadoop.hive.ql.io.avro.AvroContainerOutputFormat'

LOCATION 'hdfs:///user/hive/warehouse/categories'

TBLPROPERTIES ('avro.schema.url'='hdfs://quickstart.cloudera/user/urko/sqoop_import_categories.avsc');

categories, customers, departments, orders, order_items, productsshow tables;

exit;

Consultar datos estructurados

• Hemos transferido datos estructurados de unabase de datos relacional a HDFS

• Lo hemos transformado en formato Avro

• Hemos creado tablas HIVE, preparadas para serconsultadas mediante Impala o consultas porlotes de HIVE.

• Vamos a consultarlos!

Ejercicio 3: Consultar losdatos

63

Consultar datos estructurados

Ejercicio 3: Consultar los datos

Mostrar las categorias de producto más popularesselect 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;

65

Ejercicio 3: Consultar losdatos

66

Ejercicio 3: Consultar losdatos

67

Ejercicio 3: Consultar losdatos

68

Ejercicio 3: Consultar los datos

Mostrar los 10 productos más satisfactoriosselect 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;

69

Ejercicio 3: Consultar losdatos

70

Ejercicio 4: El valor del Big Data

El valor del Big Data

• Hasta ahora, el resultado no es nuevo: SQL!

• Pregunta de negocio: “son los productos másvistos los más vendidos?”

• Devemos correlacionar datos estructurados con otros no estructurados:

• Usando Web logs, podremos ver los productos másvistos

• Para ingerir flujos de clicks: Flume

El valor del Big Data

• Flume: un framework de tiempo real y escalablepara encaminar, filtrar, agregar, y realizarpequeñas operaciones sobre datos

• Para este ejercicio, ya tenemos recopilados un conjunto de datos de acceso web en /opt/examples/log_data/access.log.2

El valor del Big Data

Ejercicio 4: El valor del Big Data

• Cargar los datos de log a HDFS:

[cloudera@quickstart ~]$ hadoop fs -copyFromLocal /opt/examples/log_files/access.log.2

/user/hive/warehouse/original_access_logs

• Entrar en HIVE:

[cloudera@quickstart ~]$ hive

75

Ejercicio 4: El valor del Big Data

• Crear una tabla en Hive, y consultarla mediante Impala o Hive

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';

76

Ejercicio 4: El valor del Big Data

• Crear una tabla en Hive, y consultarla mediante Impala o Hive

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)

ROW FORMAT DELIMITED FIELDS TERMINATED BY ','

LOCATION '/user/hive/warehouse/tokenized_access_logs';

ADD JAR /usr/lib/hive/lib/hive-contrib.jar;

INSERT OVERWRITE TABLE tokenized_access_logs SELECT * FROM intermediate_access_logs;

Exit;

77

El valor del Big Data

Uno de los productos másvistos no está entre los máscomprados!!

El valor del Big Data

• La correlación entre ambos datos permitiódetectar un error en el precio mostrado para el producto “Adidas Kids’ RG III Mid Football Cleat”. Una vez corregido, las ventas del productoempezaron a crecer rápidamente ;)

Ejercicio 5: Advanced Analytics

Advanced Analytics

• Pregunta de negocio: ¿Qué productos se suelencomprar de forma conjunta?

• La respuesta podría ayudar a:• Optimizar campañas de marketing, posicionando

productos de forma conjunta

• Podrían utilizarlo para mejorar ventas de productosmenos vistos

• Una herramienta que permite realizar análisisrápidos sobre relaciones de objetos es Apache Spark

Ejercicio 5: Advanced Analytics

• Apache Spark

86

Ejercicio 5: Advanced Analytics

• Generamos un trabajo en Spark

[cloudera@quickstart ~]$ spark-shell --jars

/usr/lib/avro/avro-mapred.jar --conf

spark.serializer=org.apache.spark.serializer.KryoSerializer

87

Ejercicio 5: Advanced Analytics

88

Advanced Analytics

// First we're going to import the classes we need and open some of the files

// we imported from our relational database into Hadoop with Sqoop

import org.apache.avro.generic.GenericRecord

import org.apache.avro.mapred.{AvroInputFormat, AvroWrapper}

import org.apache.hadoop.io.NullWritable

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

val order_items_path = warehouse + "order_items"

val order_items = sc.hadoopFile[AvroWrapper[GenericRecord], NullWritable, AvroInputFormat[GenericRecord]](order_items_path)

val products_path = warehouse + "products"

val products = sc.hadoopFile[AvroWrapper[GenericRecord], NullWritable, AvroInputFormat[GenericRecord]](products_path)

Advanced Analytics

// Next, we extract the fields from order_items and products that we care about

// and get a list of every product, its name and quantity, grouped by order

val orders = order_items.map { x => (

x._1.datum.get("order_item_product_id"),

(x._1.datum.get("order_item_order_id"), x._1.datum.get("order_item_quantity")))

}.join(

products.map { x => (

x._1.datum.get("product_id"),

(x._1.datum.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

)

)).groupByKey()

Advanced Analytics

// Finally, we tally how many times each combination of products appears

// together in an order, and print the 10 most common combinations.

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)

println(mostCommon.deep.mkString("\n"))

Advanced Analytics

Ejercicio 6: Buscando en un mar de datos

buscando en un mar de datos

• Los logs pueden ser una fuente imprescindible de información

• Vamos a ser capaces de explorar de forma interactiva eventos de Log Web, indexando los datos, y haciendo que puedan ser buscados

• Vamos a extender Apache Flume anterior, de modo que ademas de ingerir logs, enviará los eventos a Apache Solr para su indexación en tiempo real

buscando en un mar de datos

• Flume

• Solr• Organiza los datos de la misma manera que una BBDD SQL.

• Dada registro se llama “document”, y contiene campos definidos en un esquema

• Un conjunto de documentos forman una colección.

• Es menos estricto en cuanto a la estructura:• Campos opcionales• Flexibilidad en las consultas de texto

Ingestión de Datos en streaming: Apache Flume

• Servicio distribuido y fiable para recopilar, agregary mover grandes volúmenes de datos (de log) de forma eficiente

• De muchas y variadas fuentes de logs a un lugarcentralizado de almacenamiento de datos (ej. Hadoop)

• Cuando los datos se generan de forma continua

• No solo logs, también sirve para p.e. datos de tráfico de red, datos generados por social media, emails, …

Ingestión de Datos en streaming: Apache Flume

• Una fuente externa envía eventos a un agenteFlume, en un formato que Flume puedareconocer.

• El evento se guarda en el canal, hasta que es consumido por el embudo (Sink)

• El Sink quita el evento del canal y lo pone en un repositorio externo o en otro origen de Flume

Ingestión de Datos en streaming: Apache Flume

• Origenes Flume:• Avro

• Thrift

• Exec (el resultado de la ejecución de un comando Unix)

• Un directorio a la espera de nuevos ficheros y sus contenidos

• Twitter

• NetCat

• Syslog

• HTTP

Ingestión de Datos en streaming: Apache Flume

• Canales Flume:• Memoria

• JDBC

• Fichero

• …

Ingestión de Datos en streaming: Apache Flume

• Sinks Flume:• HDFS

• Avro

• Thrift

• IRC

• Fichero

• Hbase

• …

Ingestión de Datos en streaming: Apache Flume

• Miramos la configuracion de flume:

cd /opt/examples/flume

buscando en un mar de datos

• Crear el índice de búsqueda:

solrctl --zk quickstart.cloudera:2181/solr instancedir --generate solr_configs

• Generar los ficheros de configuración:

</fields>

Ejercicio 6: Buscando en un mar de datos

• Cargamos la configuración (y morplines.conf)[cloudera@quickstart ~]$ cd /opt/examples/flume

[cloudera@quickstart ~]$ solrctl --zkquickstart.cloudera:2181/solr instancedir --create live_logs./solr_configs

• Creamos la colección[cloudera@quickstart ~]$ solrctl --zkquickstart.cloudera:2181/solr collection --create live_logs -s 1

• Arrancamos el agente flume[cloudera@quickstart ~]$ start_logs

[cloudera@quickstart ~]$ flume-ng agent --conf /opt/examples/flume/conf --conf-file /opt/examples/flume/conf/flume.conf --name agent1 -Dflume.root.logger=DEBUG,INFO,console

104

buscando en un mar de datos

Ejercicio 6: Buscando en un mar de datos

108

buscando en un mar de datos

Ejercicio 7: Creando un panel de mandos

Creando un panel de mandos

• Las búsquedas son rápidas, pero aún es fácilperdernos

• Vamos a crear un panel de mandos para analizarel comportamiento de las visitas Web en tiemporeal.