• 測試資料

    # lixiang_list.txt(小表，可以在map端載入到記憶體中)
    #立項ID 立項名稱
    1800 心願券測試003
    1801 fw心願單
    1802 wtest心願券0524
    1803 HW心願單01
    1804 心願券測試006
    1805 心願券測試007
    1806 心願券測試008


    # order_list.txt （大表）
    #訂單編號            使用者手機號  訂單金額   建立時間 立項ID  使用者ID
    18050411131170468193 18511745550 0.01 1525403591
 
 1768 6502216546
    18050411131741948542 18511745550 0.01 1525403597 1768 6502216546
    18050411132051347006 18511745550 0.01 1525403600 1768 6502216546
    18050411132624157487 18511745550 0.01 1525403606 1768 6502216546


    #連線結果
    立項ID 立項名稱 訂單編號 訂單金額 使用者手機號 使用者ID 建立時間

• 思路

  提交作業的時候，將小表文件(lixiang_list.txt)檔案以快取的形式上提交到HDFS上，然後在每個map中先將lixiang_list.txt中的資料載入到記憶體，然後根據接收到的大表的資料，判斷是否需要進行連線。

• 例項程式碼：

  
  #work.bash
  
  
  #! /bin/bash
  
  
  
  #定義map和reduce的目錄
  
  export WORK_PATH=/var/tmp/map_join
  
  #執行stream的jar包地址
  
  stream=$HADOOP_HOME/share/hadoop/tools/lib/hadoop-streaming-2.7.5.jar
  
  #資料輸入目錄
  
  input=/lcy/map_join/input
  
  #輸出結果目錄
  
  output=/lcy/map_join/output
  
  
  #刪除mr輸出目錄
  
  if $HADOOP_HOME/bin/hdfs dfs -test -d $output
  then
   
  
      $HADOOP_HOME/bin/hdfs dfs -rm -r $output
  fi
  
  
  #執行mapreduce程式
  
  $HADOOP_HOME/bin/hadoop jar $stream \
  -D mapreduce.job.reduces=3 \
  -D num.key.fields.for.partition=1  \
  -D stream.num.map.output.key.fields=1 \
  -files $WORK_PATH/lixiang_list.txt \
  -input $input/* \
  -output $output \
  -mapper "python mr.py mapper" \
  -file $WORK_PATH/mr.py

  mapreduce檔案

  
  #! /usr/bin/env python
  
  
  # coding:utf-8
  
  
  import sys
  import os
  
  dct_lx = {}
  
  def read_input(file,sepr=' '):
      for line in file:
          data = line.split(sepr)
          yield data
  
  
  def read_lx_data():
      global dct_lx
      with open('lixiang_list.txt','r') as f:
          for line in f:
              line = line.split(' ',1)
              dct_lx[line[0].strip()] = line[1].strip()
  
  def mapper():
      global dct_lx
      filepath = os.environ['map_input_file']
      filename = os.path.split(filepath)[1]
      lines = read_input(sys.stdin)
      for data in lines:
          if "order_list.txt" == filename:
              if len(data) != 6:
                  continue
              if data[4] in dct_lx:
                  print "%s\t%s\t%s\t%s\t%s\t%s\t%s" % (data[4],dct_lx[data[4]],data[0],data[2],data[1],data[5].strip('\n'),data[3])
  
  if __name__ == '__main__':
      d = {"mapper":mapper}
      if sys.argv[1] in d:
          #載入小表資料
          read_lx_data()
          d[sys.argv[1]]()