Emr hive barcamp 2012

Hadoop y Hive en AWS

Cómo analizar terabytes de datos y no morir en el intento

domingo 4 de noviembre de 2012

Quién soy yo?

• Ezequiel Golub

• Desarrollador en Smowtion

• Hace más de un año que implementamos Hadoop


Qué es Hadoop?

• Framework para aplicaciones distribuidas en miles de nodos

• Ahora familia de productos, pero en su core son 2:

• HDFS

• Map/Reduce

• Proyecto de apache.org


Limitaciones de RDBMS

• Datasets de petabytes

• no escalan horizontalmente

• Escalar horizontalmente = chau ACID

• Particionar y shardear es dificil.

• Rendimiento de lecturas o escrituras: Elegí una.

• Usar hardware especializado = $$$!!


Por qué Hadoop?

• Da respuesta las limitaciones de las RDBMS

• Escalabilidad horizontal

• Confiabilidad

• Apoyo de comunidad

• Map/Reduce

• Orientado a procesos batch para datos “grandes”

• AWS


Quienes lo usan?domingo 4 de noviembre de 2012

Cuándo usar Hadoop?

• Demasiados datos para que entren en algun tipo de datastore comodamente

• Datos desestructurados

• Y necesitas analizarlos!

• Y necesitas alta disponibilidad!

• Suficientes datos para que analizarlos en una sola maquina sea dificil


HDFS

• Sistema de archivos distribuido, redundante y con alta disponibilidad.

• Soporta archivos muy grandes.

• Pensado para commodity hardware

• Acceso streaming de baja latencia y alta transferencia.


HDFS - Ventajas

• Tolerancia a los fallos

• Autoregenerativo

• Escalable

• Confiabilidad

• Soporte


Map/Reduce

• Paradigma de programación distribuida

• Basada en un paper de Google (http://bit.ly/gXZbsk)

• Modelada a partir de las ideas de programación funcional

• Distribuible en N nodos

• map() -> reduce()

• La etapa de reduce se inicia cuando todos los mappers terminan.


http://bit.ly/gXZbsk




Map/Reduce

• Dos pasos secuenciales

• Map: Toma todas las lineas de un input, y por cada una, las procesa y devuelve un par de key valor

• Reduce: Recibe secuencialmente un key valor, los procesa y emite un resultado (ej: otro k-v)


import hadoopy

def mapper(key, value): for word in value.split(): yield word, 1

def reducer(key, values): accum = 0 for count in values: accum += int(count) yield key, accum

if __name__ == "__main__": hadoopy.run(mapper, reducer, doc=__doc__)

M/R: WC en Python

$ echo "a b a a b c" | python wc.py map | sort | python wc.py reducea 3b 2c 1

wc.py

Probandolo localmente!

** usando el modulo Hadoopy para usar python con Hadoop!


Ejemplo: wordcount

Muchos archivos

mapper

mapper

mapper

mapper<arch2.parte2>

<arch2.parte1>

<arch1.part2>

<arch1.part1>{‘foo’:12,‘bar’:13,‘baz’:19}

{‘foo’:33,‘bar’:23,‘baz’:42}

{‘foo’:1,‘bar’:0,‘baz’:99}

{‘foo’:55,‘bar’:43,‘baz’:65}

Reducer

Reducer

Reducer

Ordenar y

agrupar por key

{‘foo’:[12,33,1,55]}

{‘bar’:[13,23,0,43]}

{‘baz’:[19,42,99,65]}

count()

count()

count()

count()

sum()

sum()

sum()

{‘foo’:101,‘bar’:79,‘baz’:218}


M/R en Hadoop

• Las tareas de M/R son entregadas como un “job”

• Los “jobs” se asignan a una cola especifica de trabajo

• Los jobs son “rack-aware”: aprovecharse del “data locality”

• Cada “job” es divido en N “tasks”

• Streaming: No usar Java para M/R

• Las tareas que fallen son reiniciadas automaticamente.


M/R: Implementación

• M/R es excelente para los problemas donde los ‘sub-problemas’ no son interdependientes

• Nada se comparte entre mappers y reducers, ni siquiera si corren en el mismo nodo

• X ej: La salida de un mapper no puede depender de la salida o comunicación con otro mapper


Amazon Web Services

• Crear servidores virtuales dinamicamente.

• Pagas solo por las horas que usas

• Podes ‘apagar’ y ‘prender’ maquinas en cualquier momento


AWS: Servicios clave

• EC2: Servidores virtuales

• S3: Servicio de data-storage

• EMR: Hadoop on demand


Elastic Map Reduce

• Proveen un cluster configurado con los ultimos patches + un poco de magia extra

• Interactua nativamente con S3

• Podes agregar o sacar servers del cluster dinamicamente


Elastic Map Reduce

• Al crear el server, se lo puede acceder igual que cualquier otro server virtual

• Bootstrap actions permiten customizar el master o slaves

• Los cluster pueden ser persistente o efimeros

• 2 instance groups: CORE y TASK








Elastic Map Reduce

Ejemplos desde la consola:

# Bajar el instance group TASK a cero instancias

shell# elastic-mapreduce --jobflow j-XXXXX --modify-instance-group TASK --instance-count 0


Hive: Qué es?

• SQL sobre hadoop

• Traducir Pseudosql a tareas Map/Reduce

• Maneja datos a escala “Hadoop”

• Puede usar logfiles como source data

• WC en Hive: • SELECT COUNT(*) as freq, word FROM words GROUP by word;


Access log - crear tablaCREATE EXTERNAL TABLE access_log(ip STRING, host STRING, finishtime STRING, uri STRING, return_code STRING,bytes STRING, referrer STRING, user_agent STRING, quality STRING)comment 'data logs uploaded from servers'PARTITIONED by ( year string, month string, day string, hour string, minute string)ROW FORMAT SERDE 'org.apache.hadoop.hive.contrib.serde2.RegexSerDe'WITH SERDEPROPERTIES ( "input.regex" = '(.+?) (.+?) - (.+) \"(.+?)\" ([0-9]+) ([0-9]+) \"(.+)\" \"(.+)\" \"(.+)\"', "output.format.string" = "%1$s %2$s %3$s %4$s %5$s %6$s %7$s %8$s %9$s")STORED AS TEXTFILELOCATION 's3n://somebucket/acceslog_dir/';


http://mapreduce.in/sessions/

http://mapreduce.in/sessions/

Ejemplo con CSVCREATE EXTERNAL TABLE page_view( viewTime INT, userid BIGINT, url STRING, referrer_url STRING, ip STRING COMMENT 'IP Address of the User', country STRING COMMENT 'country of origination')ROW FORMAT DELIMITED FIELDS TERMINATED BY ','STORED AS TEXTFILELOCATION 's3n://somebucket/page_views';


Queries# Conteo de hits x url para Nov 2012SELECT count(*) as count, uri FROM access_logWHERE year = 2012 and month = 11GROUP by uri ORDER by uri DESC;

# subqueries # tablas: users, action_video y action_comment # salida: una lista con todas las acciones de todos los usuarios y sus fechas

SELECT u.id, actions.name, actions.date FROM ( SELECT av.uid AS uid, ‘video publicado’ as name FROM action_video av WHERE av.date = '2008-06-03' UNION ALL SELECT ac.uid AS uid, ‘comentario publicado’ as name FROM action_comment ac WHERE ac.date = '2008-06-03' ) actions JOIN users u ON (u.id = actions.uid)

# insertar la salida del query en s3INSERT OVERWRITE DIRECTORY ‘s3n://somebucket/somedir’SELECT * FROM access_log where uri = ‘http://example.com’


http://example.com

http://example.com

Limitaciones

• No implementa todo el SQL.

• Algunas cosas se hacen de manera rara (group by)

• Miles de archivos de salida.

• Map/Reduce es inherementente lento

• Puede ser adictivo


Tips

• Particionar las tablas (s3://bucket/access/y=2012/m=11/d=4/h=1)

• Crear sets de prueba con pocos datos

• Crear varias tablas para distintos queries

• Tabla => Query => Tabla => Query

• Borrar los datos que no usas en S3!


Precios?

• Cluster tipico

• Master: small (u$s 0.09/hora) x 1

• Slaves: c1.medium u$s 0.2/hora x 2

• Total: u$s 0.49 x hora

• S3: u$s 0.125 x gb (muy aprox)


En Smowtion?

• 70 gb x dia entre datos propios y de partners

• Hive funciona como datawarehouse y como parte del pipeline de datos:


Cómo seguir?

• Cloudera.com

• Hadoop.apache.org

• http://developer.yahoo.com/hadoop/tutorial/

• http://aws.amazon.com/developertools/2264 (AWS command line Ruby Client)

• https://cwiki.apache.org/confluence/display/Hive/LanguageManual (hive lang manual)


http://developer.yahoo.com/hadoop/tutorial/

http://developer.yahoo.com/hadoop/tutorial/

http://aws.amazon.com/developertools/2264

http://aws.amazon.com/developertools/2264

https://cwiki.apache.org/confluence/display/Hive/LanguageManual




Preguntas?

• @ezegolub

• [email protected]


Documents

Emr hive barcamp 2012