3.2.1 什么是MapReduce

• 源于Google的MapReduce論文(2004年12月)
• Hadoop的MapReduce是Google論文的開源實現(xiàn)
• MapReduce優(yōu)點: 海量數(shù)據(jù)離線處理&易開發(fā)
• MapReduce缺點: 實時流式計算

3.2.2 MapReduce編程模型

• MapReduce分而治之的思想

  • 數(shù)錢實例：一堆鈔票，各種面值分別是多少
    • 單點策略
      • 一個人數(shù)所有的鈔票，數(shù)出各種面值有多少張
    • 分治策略
      • 每個人分得一堆鈔票，數(shù)出各種面值有多少張
      • 匯總，每個人負(fù)責(zé)統(tǒng)計一種面值
    • 解決數(shù)據(jù)可以切割進(jìn)行計算的應(yīng)用

• MapReduce編程分Map和Reduce階段

  • 將作業(yè)拆分成Map階段和Reduce階段
  • Map階段 Map Tasks 分：把復(fù)雜的問題分解為若干"簡單的任務(wù)"
  • Reduce階段: Reduce Tasks 合：reduce

• MapReduce編程執(zhí)行步驟

  • 準(zhǔn)備MapReduce的輸入數(shù)據(jù)
  • 準(zhǔn)備Mapper數(shù)據(jù)
  • Shuffle
  • Reduce處理
  • 結(jié)果輸出

• 編程模型

  • 借鑒函數(shù)式編程方式

  • 用戶只需要實現(xiàn)兩個函數(shù)接口：

    • Map(in_key,in_value)

      —>(out_key,intermediate_value) list

    • Reduce(out_key,intermediate_value) list

      —>out_value list

  • Word Count 詞頻統(tǒng)計案例

    

3.2.3 Hadoop Streaming 實現(xiàn)wordcount （實驗 了解）

• Mapper

  import sys
  
  #輸入為標(biāo)準(zhǔn)輸入stdin
  for line in sys.stdin:
      #刪除開頭和結(jié)尾的空行
      line = line.strip()
      #以默認(rèn)空格分隔單詞到words列表
      words = line.split()
      for word in words:
          #輸出所有單詞，格式為“單詞 1”以便作為Reduce的輸入
          print("%s %s"%(word,1))
  

• Reducer

  import sys
  
  current_word = None
  current_count = 0
  word = None
  
  #獲取標(biāo)準(zhǔn)輸入，即mapper.py的標(biāo)準(zhǔn)輸出
  for line in sys.stdin:
      #刪除開頭和結(jié)尾的空行
      line = line.strip()
  
      #解析mapper.py輸出作為程序的輸入，以tab作為分隔符
      word,count = line.split()
  
      #轉(zhuǎn)換count從字符型到整型
      try:
          count = int(count)
      except ValueError:
          #count非數(shù)字時，忽略此行
          continue
  
      #要求mapper.py的輸出做排序（sort）操作，以便對連續(xù)的word做判斷
      if current_word == word:
          current_count += count
      else :
          #出現(xiàn)了一個新詞
          #輸出當(dāng)前word統(tǒng)計結(jié)果到標(biāo)準(zhǔn)輸出
          if current_word :
              print('%s\t%s' % (current_word,current_count))
          #開始對新詞的統(tǒng)計
          current_count = count
          current_word = word
  
  #輸出最后一個word統(tǒng)計
  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的輸出數(shù)據(jù)進(jìn)行字典排序

• 通過Hadoop Streaming 提交作業(yè)到Hadoop集群

  STREAM_JAR_PATH="/root/bigdata/hadoop/share/hadoop/tools/lib/hadoop-streaming-2.9.1.jar"    # hadoop streaming jar包所在位置
  INPUT_FILE_PATH_1="/The_Man_of_Property.txt"  #要進(jìn)行詞頻統(tǒng)計的文檔在hdfs中的路徑
  OUTPUT_PATH="/output"                         #MR作業(yè)后結(jié)果的存放路徑
  
  hadoop fs -rm -r -skipTrash $OUTPUT_PATH    # 輸出路徑如果之前存在 先刪掉否則會報錯
  
  hadoop jar $STREAM_JAR_PATH \   
  		-input $INPUT_FILE_PATH_1 \ # 指定輸入文件位置
  		-output $OUTPUT_PATH \      #指定輸出結(jié)果位置
  		-mapper "python map.py" \   #指定mapper執(zhí)行的程序
  		-reducer "python red.py" \  # 指定reduce階段執(zhí)行的程序
  		-file ./map.py \            # 通過-file 把python源文件分發(fā)到集群的每一臺機(jī)器上  
  		-file ./red.py
  

• 到Hadoop集群查看運(yùn)行結(jié)果

  文章來源地址http://www.zghlxwxcb.cn/news/detail-515500.html

到了這里，關(guān)于分布式處理框架 MapReduce的文章就介紹完了。如果您還想了解更多內(nèi)容，請在右上角搜索TOY模板網(wǎng)以前的文章或繼續(xù)瀏覽下面的相關(guān)文章，希望大家以后多多支持TOY模板網(wǎng)！