## 3.3 Hadoop Streaming 实现 word count

需求：统计单词数量

- Mapper

  ```python
  import sys
  
  #输入为标准输入stdin
  for line in sys.stdin:
      #删除开头和结尾的空行
      line = line.strip()
      #以默认空格分隔单词到words列表
      words = line.split()
      for word in words:
          #输出所有单词，格式为“单词 1”以便作为Reduce的输入
          print("%s %s"%(word,1))
  ```

- Reducer

  ```python
  import sys
  
  current_word = None
  current_count = 0
  word = None
  
  #获取标准输入，即mapper.py的标准输出
  for line in sys.stdin:
      #删除开头和结尾的空行
      line = line.strip()
  
      #解析mapper.py输出作为程序的输入，以tab作为分隔符
      word,count = line.split()
  
      #转换count从字符型到整型
      try:
          count = int(count)
      except ValueError:
          #count非数字时，忽略此行
          continue
  
      #要求mapper.py的输出做排序（sort）操作，以便对连续的word做判断
      if current_word == word:
          current_count += count
      else :
          #出现了一个新词
          #输出当前word统计结果到标准输出
          if current_word :
              print('%s\t%s' % (current_word,current_count))
          #开始对新词的统计
          current_count = count
          current_word = word
  
  #输出最后一个word统计
  if current_word == word:
      print("%s\t%s"% (current_word,current_count))
  ```

  cat xxx.txt|python3 map.py|sort|python3 red.py

  得到最终的输出

  注：hadoop-streaming会主动将map的输出数据进行字典排序

- 通过Hadoop Streaming 提交作业到Hadoop集群

  ```shell
  HADOOP_CMD="/hadoop_home/bin/hadoop" #hadoop安装路径
  STREAM_JAR_PATH="/hadoop_home/share/hadoop/tools/lib/hadoop-streaming-2.6.0-cdh5.7.0.jar" # hadoop streaming jar包所在位置
  
  INPUT_FILE_PATH="/xxx.txt"  #要进行词频统计的文档在hdfs中的路径
  OUTPUT_PATH="/output"                         #MR作业后结果的存放路径
  
  $HADOOP_CMD fs -rmr -skipTrash $OUTPUT_PATH # 输出路径如果之前存在 先删掉否则会报错
  
  $HADOOP_CMD jar $STREAM_JAR_PATH\   
   -input $INPUT_FILE_PATH\ # 指定输入文件位置
   -output $OUTPUT_PATH\      #指定输出结果位置
   -mapper "python map.py"\   #指定mapper执行的程序
   -reducer "python red.py"\  # 指定reduce阶段执行的程序
   -file ./map.py\            # 通过-file 把python源文件分发到集群的每一台机器上  
   -file ./red.py
  ```

- 到Hadoop集群查看运行结果

![](img/mr_result.png)