Cómo escribir un programa MapReduce simple para Hadoop usando Python

En este ejemplo, le presentaré cómo usar Python para escribir un programa MapReduce simple para Hadoop

.

Aunque el marco de Hadoop está escrito en Java, todavía necesitamos usar lenguajes como C y Python para implementar programas de Hadoop. Aunque el programa de muestra proporcionado por el sitio web oficial de Hadoop está escrito en Jython y empaquetado en un archivo Jar, esto obviamente causa inconvenientes. De hecho, no es necesario implementarlo de esta manera. Podemos usar Python para asociarnos con Hadoop para la programación. Eche un vistazo a /src/examples/python/WordCount.py y verá de qué estoy hablando.

¿Qué queremos hacer?

Escribiremos un programa MapReduce simple usando C-Python en lugar de un programa escrito en Jython y empaquetado en un paquete jar.

Nuestro ejemplo imitará WordCount y lo implementará usando Python. El ejemplo cuenta el número de apariciones de palabras leyendo un archivo de texto. Los resultados también se muestran en forma de texto. Cada línea contiene una palabra y el número de veces que aparece, separados por tabulaciones.

Requisitos previos

Antes de escribir este programa, debe configurar un clúster de Hadoop para no quedar cegado por el trabajo posterior. Si aún no lo ha configurado, hay un breve tutorial a continuación para enseñarle cómo configurarlo en Ubuntu Linux (también aplicable a otras distribuciones de Linux y Unix)

Cómo usar Hadoop Distributed El sistema de archivos (HDFS) en Ubuntu Linux establece un clúster de Hadoop de un solo nodo

Cómo utilizar el sistema de archivos distribuido Hadoop (HDFS) para establecer un clúster de Hadoop de múltiples nodos en Ubuntu Linux

Código Python MapReduce

El truco de escribir código MapReduce en Python es que usamos HadoopStreaming para ayudarnos a pasar datos entre Map y Reduce a través de STDIN (entrada estándar) y STDOUT (salida estándar). Solo usamos el sistema de Python. .stdin para ingresar datos, usando sys .stdout genera datos. Esto se hace porque HadoopStreaming hará otras cosas por nosotros. ¡Es verdad, no lo creas!

Mapa: mapper.py

Guarde el siguiente código en /home/hadoop/mapper.py, leerá los datos de STDIN y separará las palabras en líneas. Generará una lista que mapea los relación entre palabras y ocurrencias:

Nota: asegúrese de que este script tenga permisos suficientes (chmod x /home/hadoop/mapper.py).

#!/usr/bin/env python

import sys

# la entrada proviene de STDIN (entrada estándar)

para la línea en sys.stdin:

# eliminar los espacios en blanco iniciales y finales

line = line.strip()

# dividir la línea en palabras

palabras = line.split()

# aumentar contadores

para palabra en palabras:

# escribir los resultados en STDOUT (salida estándar);

# lo que generamos aquí será la entrada para el

# paso Reducir, es decir, la entrada para reducer.py

#

# delimitado por tabulaciones; el recuento de palabras trivial es 1

print 's\\ts' (palabra, 1) En este script, el número total de apariciones de palabras no se calcula, generará "lt; wordgt; 1" rápidamente, aunque lt; wordgt; puede aparecer varias veces en la entrada, el cálculo se deja para Reducir pasos (o programas) posteriores. Por supuesto, puedes cambiar tu estilo de codificación y respetar plenamente tus hábitos.

Reducir: reducer.py

Almacena el código en /home/hadoop/reducer.py La función de este script es leer el resultado del STDIN de mapper.py. y luego Calcule el número total de apariciones de cada palabra y envíe el resultado a STDOUT.

Además, preste atención a los permisos del script: chmod x /home/hadoop/reducer.py

#!/usr/bin/env python

del operador import itemgetter

import sys

# asigna palabras a sus recuentos

word2count = {}

# la entrada proviene de STDIN

para la línea en sys.stdin:

# eliminar los espacios en blanco iniciales y finales

line = line.strip()

# analiza la entrada que obtenido de mapper.py

palabra, count = line.split('\\t', 1)

# convierte count (actualmente una cadena) a int

intente:

recuento = int(recuento)

word2count[palabra] = word2count.get(palabra, 0) recuento

excepto ValueError:

# el recuento no era un número, así que en silencio

# ignora/descarta esta línea

pasa

# ordena las palabras lexigráficamente;

#

# este paso NO es necesario, solo lo hacemos para que nuestro

# resultado final se parezca más al Hadoop oficial

# ejemplos de recuento de palabras

sorted_word2count = sorted(word2count.items(), key=itemgetter(0))

# escribir los resultados en STDOUT (salida estándar)

para palabra, cuenta en sorted_word2count:

imprime 's\\ts' (palabra, recuento)

Prueba tu código (datos de gato|mapa|ordenar|reducir)

Le recomiendo que intente probar manualmente sus scripts mapper.py y reducer.py antes de ejecutar la prueba del trabajo MapReduce para evitar obtener resultados.

Aquí hay algunas sugerencias sobre cómo probar el funcionalidad de tu Mapa y Reducir:

——————————————————————————————————— —— ——————————

\r\n

# prueba muy básica

hadoop@

ubuntu: ~$ echo "foo foo quux labs foo bar quux" /home/hadoop/mapper.py

foo 1

foo 1

quux 1

laboratorios 1

foo 1

barra 1

———————————————— ———————————————————————————————

hadoop@ubuntu:~$ echo "foo foo quux labs foo bar quux" | /home/hadoop/mapper.py | sort | /home/hadoop/reducer.py

bar 1

foo 3

labs 1

———————————————————————————————————— ————

# usando uno de los libros electrónicos como entrada de ejemplo

# (ver más abajo dónde conseguir los libros electrónicos)

hadoop@ubuntu：~$ cat /tmp/gutenberg/ 20417-8.txt | home/hadoop/mapper.py

El 1

Proyecto 1

Gutenberg 1

Libro electrónico 1

de 1

[...]

(ya entiendes la idea)

quux 2

quux 1

——————————————————————————————————— ——————— —————

Ejecución de scripts Python en la plataforma Hadoop

Para este ejemplo, necesitaremos tres libros electrónicos:

The Outline of Science, Vol. 1 (de 4) de J. Arthur Thomson\r\n

Los cuadernos de Leonardo Da Vinci\r\n

Ulises de James Joyce

Descárguelos y use la codificación us-ascii para almacenar los archivos descomprimidos en un directorio temporal, como /tmp/gutenberg.

hadoop@ubuntu: ~$ ls -l / tmp/gutenberg/

total 3592

-rw-r--r-- 1 hadoop hadoop 674425 2007-01-22 12:56 20417-8.txt

-rw-r --r-- 1 hadoop hadoop 1423808 2006-08-03 16:36 7ldvc10.txt

-rw-r--r-- 1 hadoop hadoop 1561677 2004-11 -26 09:48 ulyss12. txt

hadoop@ub

untu:~$

Copiar datos locales a HDFS

Antes de ejecutar el trabajo MapReduce, necesitamos copiar los archivos locales a HDFS:

hadoop@ ubuntu :/usr/local/hadoop$ bin/hadoop dfs -copyFromLocal /tmp/gutenberg gutenberg

hadoop@ubuntu:/usr/local/hadoop$ bin/hadoop dfs -ls

Se encontraron 1 artículos

/user/hadoop/gutenberg lt;dirgt;

hadoop@ubuntu:/usr/local/hadoop$ bin/hadoop dfs -ls gutenberg

Se encontraron 3 elementos

/user/hadoop/gutenberg/20417-8.txt lt;r 1gt; 674425

/user/hadoop/gutenberg/7ldvc10.txt lt; 1gt; 1423808

/user/hadoop/gutenberg/ulyss12.txt lt; r 1gt; 1561677

Ejecutar el trabajo MapReduce

Ahora todo está listo. ejecutará un trabajo Python MapReduce en un clúster de Hadoop. Como dije anteriormente, usamos HadoopStreaming para ayudarnos a transferir datos entre Map y Reduce y a través de STDIN y STDOUT para entrada y salida estandarizadas.

hadoop@ubuntu:/usr/local/hadoop$ bin/hadoop jar contrib/streaming/hadoop-0.19.1-streaming.jar

-mapper /home/hadoop/mapper .py -reducer /home/hadoop/reducer.py -input gutenberg/*

-output gutenberg-output

Durante la operación, si desea cambiar algunas configuraciones de Hadoop, como Para aumentar el número de tareas de Reducir, puede utilizar la opción "-jobconf":

hadoop@ubuntu:/usr/local/hadoop$ bin/hadoop jar contrib/streaming/hadoop-0.19.1 -streaming.jar

-jobconf mapred.reduce.tasks=16 -mapper ...

Una nota importante sobre Hadoop no respeta mapred.map.tasks

Esta tarea leerá los archivos gutenberg en el directorio HDFS y los procesará, almacenando los resultados en un archivo de resultados separado en el directorio gutenberg-output en el directorio HDFS.

Los resultados de la ejecución anterior son los siguientes:

hadoop@ubuntu:/usr/local/hadoop$ bin/hadoop jar contrib/streaming/hadoop-0.19.1-streaming .jar

p>

-mapper /home/hadoop/mapper.py -reducer /home/hadoop/reducer.py -input gutenberg/*

-output gutenberg- salida

additionalConfSpec_ ：null

null=@@@userJobConfProps_.get(stream.shipped.hadoopstreaming

packageJobJar: [/usr/local/hadoop-datastore /hadoop-hadoop/hadoop-unjar54543/ ]

[] /tmp/streamjob54544.jar tmpDir=null

[...] INFORMACIÓN mapred.FileInputFormat: Rutas de entrada totales para procesar : 7

[...] INFORMACIÓN streaming.StreamJob: getLocalDirs(): [/usr/local/hadoop-datastore/hadoop-hadoop/mapred/local]

[. ..] INFO streaming.StreamJob: Trabajo en ejecución: job_200803031615_0021

[...]

[...] INFO streaming.StreamJob: mapa 0 reducir 0

[...] INFO streaming .StreamJob: mapa 43 reducir 0

[...] INFO streaming.StreamJob: mapa 86 reducir 0

[...] INFO streaming.StreamJob: mapa 100 reducir 0

[...] INFO streaming.StreamJob: mapa 100 reducir 33

[...] INFO streaming.StreamJob: mapa 100 reducir 70

[..] INFO streaming.StreamJob： mapa 100 reducir 77

[…] INFO streaming.StreamJob： mapa 100 reducir 100

[…] INFO streaming.StreamJob: Trabajo completo: job_200803031615_0021

[...] INFO streaming.StreamJob: Salida: gutenberg-output hadoop@ubuntu:/usr/local/hadoop$

As puedes ver el resultado anterior

Como resultado, Hadoop también proporciona una interfaz WEB básica para mostrar información y resultados estadísticos.

Cuando el clúster de Hadoop se está ejecutando, puede utilizar el navegador para acceder a http://localhost:50030/, como se muestra en la figura:

Compruebe si los resultados se generan y almacenado en el directorio HDFS gutenberg-output:

hadoop@ubuntu:/usr/local/hadoop$ bin/hadoop dfs -ls gutenberg-output

Se encontraron 1 elementos

/ usuario/hadoop/gutenberg-output/part-00000 lt; r 1gt; 903193 2007-09-21 13:00

hadoop@ubuntu:/usr/local/hadoop$

Puede usar el comando dfs -cat para verificar el directorio de archivos

hadoop@ubuntu:/usr/local/hadoop$ bin/hadoop dfs -cat gutenberg-output/part-00000

"(Lo )cra" 1

"1490 1

"1498," 1

"35" 1

"40," 1

"A 2

"TAL CUAL". 2

"A_ 1

" Absoluti 1

[...]

hadoop@ubuntu:/usr/local/hadoop$

Tenga en cuenta la salida, el símbolo (") en el Hadoop no inserta el resultado anterior.

Reimpreso solo como referencia, los derechos de autor pertenecen al autor original. Le deseo una vida feliz, acéptelo si está satisfecho.