Développement Python Série MapReduce Démo WordCount

Nous savons que MapReduce est le cœur de l'éléphant Hadoop. Dans Hadoop, le cœur du traitement des données est le modèle de programmation MapReduce. Un Map/Reduce divise généralement l'ensemble de données d'entrée en plusieurs blocs de données indépendants, qui sont traités de manière complètement parallèle par des tâches de mappage (tâche) . Le framework triera d'abord la sortie de la carte, puis saisira les résultats dans la tâche de réduction. Généralement, les entrées et sorties d'une tâche sont stockées dans le système de fichiers. Par conséquent, notre centre de programmation est principalement l’étape mappeur et l’étape réducteur.

Développons un programme MapReduce à partir de zéro et exécutons-le sur un cluster Hadoop.
code du mappeur map.py :

 import sys    
    for line in sys.stdin:
        word_list = line.strip().split(' ')    
        for word in word_list:            print '\t'.join([word.strip(), str(1)])

Afficher le code

code du réducteur réduire.py :

 import sys
    
    cur_word = None
    sum = 0    
    for line in sys.stdin:
        ss = line.strip().split('\t')        
        if len(ss) < 2:            continue
    
        word = ss[0].strip()
        count = ss[1].strip()    
        if cur_word == None:
            cur_word = word    
        if cur_word != word:            print &#39;\t&#39;.join([cur_word, str(sum)])
            cur_word = word
            sum = 0
        
        sum += int(count)    
    print &#39;\t&#39;.join([cur_word, str(sum)])
    sum = 0

Afficher le code

Fichier de ressources src.txt (pour les tests, n'oubliez pas de télécharger sur HDFS lors de l'exécution dans le cluster) :

hello    
    ni hao ni haoni hao ni hao ni hao ni hao ni hao ni hao ni hao ni hao ni hao ni hao ni hao ao ni haoni hao ni hao ni hao ni hao ni hao ni hao ni hao ni hao ni hao ni hao ni haoao ni haoni hao ni hao ni hao ni hao ni hao ni hao ni hao ni hao ni hao ni hao ni hao
    Dad would get out his mandolin and play for the family
    Dad loved to play the mandolin for his family he knew we enjoyed singing
    I had to mature into a man and have children of my own before I realized how much he had sacrificed
    I had to,mature into a man and,have children of my own before.I realized how much he had sacrificed

Afficher le code

Déboguer d'abord localement pour voir si le résultat est correct Entrez la commande suivante :

cat src.txt | python map.py | sort -k 1 | python reduce.py

Sortie dans. la ligne de commande Résultat :

a    2
    and    2
    and,have    1
    ao    1
    before    1
    before.I    1
    children    2
    Dad    2
    enjoyed    1
    family    2
    for    2
    get    1
    had    4
    hao    33
    haoao    1
    haoni    3
    have    1
    he    3
    hello    1
    his    2
    how    2
    I    3
    into    2
    knew    1
    loved    1
    man    2
    mandolin    2
    mature    1
    much    2
    my    2
    ni    34
    of    2
    out    1
    own    2
    play    2
    realized    2
    sacrificed    2
    singing    1
    the    2
    to    2
    to,mature    1
    we    1
    would    1

Afficher le code

Le débogage local a été trouvé grâce au débogage et le code est OK. Jetez-le sur le cluster et exécutez. Pour plus de commodité, j'ai écrit un script spécial run.sh pour libérer la main-d'œuvre.

HADOOP_CMD="/home/hadoop/hadoop/bin/hadoop"
    STREAM_JAR_PATH="/home/hadoop/hadoop/contrib/streaming/hadoop-streaming-1.2.1.jar"
    
    INPUT_FILE_PATH="/home/input/src.txt"
    OUTPUT_PATH="/home/output"
    
    $HADOOP_CMD fs -rmr  $OUTPUT_PATH 
    
    $HADOOP_CMD jar $STREAM_JAR_PATH \        -input $INPUT_FILE_PATH \        -output $OUTPUT_PATH \        
    -mapper "python map.py" \        -reducer "python reduce.py" \        -file ./map.py \        -file ./reduce.py

Analysons le script ci-dessous :

　HADOOP_CMD： hadoop的bin的路径
    STREAM_JAR_PATH：streaming jar包的路径
    INPUT_FILE_PATH：hadoop集群上的资源输入路径
    OUTPUT_PATH：hadoop集群上的结果输出路径。（注意：这个目录不应该存在的，因此在脚本加了先删除这个目录。**注意****注意****注意**：若是第一次执行，没有这个目录，会报错的。可以先手动新建一个新的output目录。）
    $HADOOP_CMD fs -rmr  $OUTPUT_PATH
    
    $HADOOP_CMD jar $STREAM_JAR_PATH \        -input $INPUT_FILE_PATH \        -output $OUTPUT_PATH \       
     -mapper "python map.py" \        -reducer "python reduce.py" \       
      -file ./map.py \        -file ./reduce.py                 
      #这里固定格式，指定输入，输出的路径；指定mapper，reducer的文件；
      #并分发mapper，reducer角色的我们用户写的代码文件，因为集群其他的节点还没有mapper、reducer的可执行文件。

Entrez la commande suivante pour afficher les enregistrements de sortie après la phase de réduction :

cat src.txt | python map.py | sort -k 1 | python reduce.py | wc -l
命令行中输出：43

Entrez : master:50030 dans le navigateur pour afficher les détails de la tâche.

Kind    % Complete    Num Tasks    Pending    Running    Complete    Killed     Failed/Killed Task Attempts
map       100.00%        2            0        0        2            0            0 / 0
reduce    100.00%        1            0        0        1            0            0 / 0

Voir ceci dans Map-Reduce Framework.

Counter                    　　Map    Reduce    Total
Reduce output records    0    　　0       　　 43

La preuve que l'ensemble du processus a été réussi. Le développement du premier programme hadoop est terminé.

Ce qui précède est le contenu détaillé de. pour plus d'informations, suivez d'autres articles connexes sur le site Web de PHP en chinois!