實現student和course資料表的join操作,以學生編號（sno）為連線欄位

• 測試資料

  • student.txt檔案

  
  #以一個空格分隔
  
  
  #學生編號   姓名
  
  
  #sno    sname
  
  01 lily
  02 tom
  03 jack
  04 rose

  • course.txt檔案

    
    #以一個空格分隔
    
    
    #學生編號    課程名  課程成績
    
    
    #sno cname grade 
    
    01 English 80
    01 Math 90
    02 English 82
    02 Math 95

  • 最終reduce端連線結果

  sno  sname   cname   grade
  01   lily    English   80
   
  
  01   lily    Math      90
  02   tom     English   82
  02   tom     Math      95

• 思路1

  按照不同的檔案輸出時對資料新增標記,然後通過配置map階段的排序欄位，保證學生資訊出現在課程成績資訊的最上面，reduce中，發現是學生資訊時，對後面接收的課程資訊進行join並輸出。步驟如下：

  1：map端有多個輸入檔案，以os.environ[‘map_input_file’]獲取輸入的檔案資訊，如果是student.txt檔案的輸入，則在輸出中增加一個’0’,是course.txt的輸入，在輸出中增加一個’1’.

  2： 以sno為key進行partation,確保將同一個學生的資訊和課程資訊分配到同一個reduce中。同時以”sno,標識” 作為map階段輸出的排序依據（以\t分隔的前兩列）,確保在reduce中同一個學生的學生資訊在課程資訊的上面（sno,0在所有sno,1的行前面）。

  3： 在reduce中，遇到0標記的行儲存sno和sname資訊，遇到1標記的行將sno和sname和當前行的課程資訊一起輸出。

  • 示例程式碼

    mapper和reducer程式碼

        #! /usr/bin/env python
        # coding:utf-8
        import os
        import
     
     sys
    
        def read_input(file,sepr=' '):
            for line in file:
                line = line.split(sepr)
                yield line
    
        def mapper():
            filepath = os.environ['map_input_file']
            filename = os.path.split(filepath)[1]
            lines = read_input(sys.stdin)
            for data in lines:
                if data[0].strip() == "":
                    continue
    
                if "student.txt" == filename:
                    if len(data) != 2:
                        continue
                    else:
                        print "%s\t%s\t%s" % (data[0],"0",data[1].strip())
                else:
                    if len(data) != 4:
                        continue
                    else:
                        print "%s\t%s\t%s\t%s" %(data[0],"1",data[2].strip(),data[3].strip())
    
        def reducer():
            sno = None
            sname = None
    
            line = read_input(sys.stdin,'\t')
            for data in line:
                if (len(data) !=3 and len(data) !=4 ) or data[0].split() == "":
                    continue
                if data[0] != sno:
                    sno = data[0]
                    if data[1] == "0":
                        sname = data[2].strip('\n')
                else:
                    cname = data[2]
                    cnum = data[3].strip('\n')
                    print "%s\t%s\t%s\t%s" % (sno,sname,cname,cnum)
    
    
    
        if __name__ == "__main__":
            d = {"mapper":mapper,"reducer":reducer}
            if sys.argv[1] in d:
                d[sys.argv[1]]()

    work.bash

    
    #! /bin/bash
    
    
    #定義map和reduce的目錄
    
    export WORK_PATH=/var/tmp/reduce_join
    
    #執行stream的jar包地址
    
    stream=$HADOOP_HOME/share/hadoop/tools/lib/hadoop-streaming-2.7.5.jar
    
    #資料輸入目錄
    
    input=/lcy/reduce_join/input
    
    #輸出結果目錄
    
    output=/lcy/reduce_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=2 \        #設定reduce個數
    -D num.key.fields.for.partition=1   \   #設定第一個欄位為分割槽欄位（預設使用\t作為map輸出分隔符）
    -D stream.num.map.output.key.fields=2 \  #設定map的輸出以前兩個欄位作為排序依據，這個配置很重要，否則reduce的輸入格式不正確
    -partitioner org.apache.hadoop.mapred.lib.KeyFieldBasedPartitioner \  #和partition引數一起使用
    -input $input/* \
    -output $output \
    -mapper "python mr.py mapper" \
    -reducer "python mr.py reducer" \
    -file $WORK_PATH/mr.py

• 思路2

  在reduce中，將學生資訊和課程成績資訊儲存到記憶體中，等遍歷完一個學生後，對兩個資料進行join操作。這種方式可以讓學生資訊在課程資訊的後面（因為經過partation和shuffle後，同一個sno的學生資訊和課程成績資訊會是連續的，要麼學生資訊在上面，要麼學生資訊在下面）步驟如下：

  1：同思路1的第一步

  2：以sno為key進行partation,確保將同一個學生的資訊和課程資訊分配到同一個reduce中。其中map端的排序可以不用設定了。

  3：reduce中，遇到0標記的行儲存sno和sname資訊，遇到1標記的行，將cname和grade組成list並追加到arr中。遇到新的sno時遍歷arr,並和sno,sname一起輸出。

  • 測試程式碼

  mapper和reducer

  
  #! /usr/bin/env python
  
  
  # coding:utf-8
  
  import os
  import sys
  
  def read_input(file,sepr=' '):
      for line in file:
          line = line.split(sepr)
          yield line
  
  def mapper():
      filepath = os.environ['map_input_file']
      filename = os.path.split(filepath)[1]
      lines = read_input(sys.stdin)
      for data in lines:
          if data[0].strip() == "":
              continue
  
          if "student.txt" == filename:
              if len(data) != 2:
                  continue
              else:
                  print "%s\t%s\t%s" % (data[0],"0",data[1].strip())
          else:
              if len(data) != 3:
                  continue
              else:
                  print "%s\t%s\t%s\t%s" %(data[0],"1",data[1].strip(),data[2].strip())
  
  def reducer():
      sno = None
      sname = None
      tpl = []
      line = read_input(sys.stdin,'\t')
      for data in line:
          if (len(data) !=3 and len(data) !=4 ) or data[0].split() == "":
              continue
  
          if data[0] != sno:
              #遇到新的sno時，進行連線輸出
              if sno != None and len(tpl) > 0:
                  for row in tpl:
                      print '%s\t%s\t%s\t%s' % (sno,sname,row[0],row[1])
                  tpl = []   #清空list，否則會變成笛卡爾積
  
              sno = data[0]
              if data[1] == "0":
                  sname = data[2].strip('\n')
              else:
                  tpl.append([data[2].strip(),data[3].strip()])
          else:
              if data[1] == "0":
                  sname = data[2].strip('\n')
              else:
                  tpl.append([data[2].strip(),data[3].strip()])
  
      if sno != None and len(tpl) > 0:
          for row in tpl:
              print '%s\t%s\t%s\t%s' % (sno,sname,row[0],row[1])
  
  
  
  if __name__ == "__main__":
      d = {"mapper":mapper,"reducer":reducer}
      if sys.argv[1] in d:
          d[sys.argv[1]]()

  work.bash

  
  #! /bin/bash
  
  
  #定義map和reduce的目錄
  
  export WORK_PATH=/var/tmp/reduce_join2
  
  #執行stream的jar包地址
  
  stream=$HADOOP_HOME/share/hadoop/tools/lib/hadoop-streaming-2.7.5.jar
  
  #資料輸入目錄
  
  input=/lcy/reduce_join2/input
  
  #輸出結果目錄
  
  output=/lcy/reduce_join2/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=2 \
  -D num.key.fields.for.partition=1   \
  -partitioner org.apache.hadoop.mapred.lib.KeyFieldBasedPartitioner \
  -input $input/* \
  -output $output \
  -mapper "python mr.py mapper" \
  -reducer "python mr.py reducer" \
  -file $WORK_PATH/mr.py