hadoop的mapreduce的join操作原理

 

1. 概述

如果我们有如下的两个文件：

person.txt(字段是id， name，addressId):

1       tom     100

2       jme     101

3       kite    102

4       jack    100

5       tim     101

address.txt(字段是id，name)：

100     Beijing

101     Shanghai

102     Guangzhou

103     Shenzhen

最后需要输出person所在的位置信息，形式如下：

100         1     tom  Beijing

100         4     jack Beijing

101         2     jme  Shanghai

101         5     tim   Shanghai

102         3     kite  Guangzhou

 

2. 实现的原理

上述两个文件，若看成两张表，则连接字段是person.addressId= address.id，此处讨论的实现方式INNER JOIN。

2.1. reduce side join的方式

 reduce side join是一种最简单的join方式，其主要思想如下：在map阶段，统一读取所有源的数据，例如，这里读取person.txt和address.txt；map的输出的key是join的字段对应的值，输出的value是每条记录的其他字段值，为了对数据来源做区分，每个value值的基础上还需要增加一个标记值（tag），例如，来自person.txt的tag设置为0，来自address.txt的tag设置为1。相比普通的mapreduce任务，join操作的map输出的value值增加了一个tag标记值。在reduce阶段，reduce函数获取key对应的value列表，将value列表根据来源的tag进行分组，最后执行笛卡尔积进行输出，即reduce阶段进行实际的连接操作。

hadoop的contrib/data_join 包org.apache.hadoop.contrib.utils.join中已经提供了此种join方式的实现，基于上述包完成的join操作的代码如下：

SampleTaggedMapOutput.java：

 

public class SampleTaggedMapOutput extends TaggedMapOutput {
  private Text data;
 
  public SampleTaggedMapOutput() {
    this.data = new Text("");
  }
 
  public SampleTaggedMapOutput(Text data) {
    this.data = data;
  }
 
  public Writable getData() {
    return data;
  }
 
  public void write(DataOutput out) throws IOException {
    this.tag.write(out);
    this.data.write(out);
  }
 
  public void readFields(DataInput in) throws IOException {
    this.tag.readFields(in);
    this.data.readFields(in);
  }
}

 

 

SampleDataJoinMapper.java：

 

public class SampleDataJoinMapper extends DataJoinMapperBase {
 
 
       @Override
       protected Text generateInputTag(String filename) {
              if (filename.contains("person")) {
                     return new Text("0");
              } else {
                     return new Text("1");
              }
       }
 
  protected Text generateGroupKey(TaggedMapOutput aRecord) {
    String line = ((Text) aRecord.getData()).toString();
    String groupKey = "";
    String[] tokens = line.split("\\t");
    if (this.inputTag.toString().equals("0")) {
           groupKey = tokens[2];
    } else {
           groupKey = tokens[0];
    }
    return new Text(groupKey);
  }
 
  protected TaggedMapOutput generateTaggedMapOutput(Object value) {
    TaggedMapOutput retv = new SampleTaggedMapOutput((Text) value);
    retv.setTag(new Text(this.inputTag));
    return retv;
  }
}
 

 

 

SampleDataJoinReducer.java:

 

public class SampleDataJoinReducer extends DataJoinReducerBase {
  /**
   *
   * @param tags
   *          a list of source tags
   * @param values
   *          a value per source
   * @return combined value derived from values of the sources
   */
  protected TaggedMapOutput combine(Object[] tags, Object[] values) {
    // eliminate rows which didnot match in one of the two tables (for INNER JOIN)
    if (tags.length < 2)
       return null; 
    String joinedStr = "";
    String [] p = null, a = null;
    Text t = (Text) tags[0];
       if (t.toString().equals("0")) {
              p =  ((Text) (((TaggedMapOutput) values[0]).getData())).toString().split("\t");
              a = ((Text) (((TaggedMapOutput) values[1]).getData())).toString().split("\t");
       } else {
              p =  ((Text) (((TaggedMapOutput) values[1]).getData())).toString().split("\t");
              a = ((Text) (((TaggedMapOutput) values[0]).getData())).toString().split("\t");
       }
             
       joinedStr = p[0] + "\t" + p[1] + "\t" + a[1];
 
    TaggedMapOutput retv = new SampleTaggedMapOutput(new Text(joinedStr));
    retv.setTag((Text) tags[0]);
   
    return retv;
  }
}

 

 

小结：reduce side join技术是灵活的，但是大部分情况它会变得效率极低。由于join直到reduce()阶段才会开始，我们将会在网络中传递shuffle所有数据（执行copy，sort等动作），然而在大多数情况下，join阶段会丢掉大多数传递的数据。因此，得到的启示有两点：

（1）如果确定是reduce-side的join，那么参与join的文件在map端尽可能先过滤掉无关的数据，例如针对特定的文件的projection/filtering，而不是传递到reduce节点后，在join时才做

（2）是否可以直接在map端就完成join操作，答案是肯定的。

2.2 map side join的方式

1. 其中一个join的文件比较小，能够完全放进内存（In-Memory Hash Join）

其实这个类似数据库中classic hash join的方式，就是一个是build input阶段，一个是probe input阶段，这里都在map端完成。build input阶段，读取小文件的数据到内存，构建一个hashtable；probe input阶段，读取大文件，通过查询hashtable实现join。

具体在hadoop中，已有相关的类支持：

（1）使用DistributedCache类，原理上就是在集群中的每个DataNode上的LocalFS都帮复制一份需要的小文件，然后你的每个Mapper进程读的就都是本地的那个文件，之后建立hashtable

（2）在mapper的类里查询hashtable，同时实现join

针对上面那个例子：mapper类的操作如下：

 

public class MapClass extends MapReduceBase implements
              Mapper {
       private Hashtable<String, String> joinData = new Hashtable<String, String>();
 
       @Override
       public void configure(JobConf conf) {
              try {
                     Path[] cacheFiles = DistributedCache.getLocalCacheFiles(conf);
                     if (cacheFiles != null && cacheFiles.length > 0) {
                            String line;
                            String[] tokens;
                            BufferedReader joinReader = new BufferedReader(new FileReader(
                                          cacheFiles[0].toString()));
                            try {
                                   while ((line = joinReader.readLine()) != null) {
                                          tokens = line.split("\\t", 2);
                                          joinData.put(tokens[0], tokens[1]);
                                   }
                            } finally {
                                   joinReader.close();
                            }
                     }
              } catch (IOException e) {
                     System.err.println("Exception reading DistributedCache: " + e);
              }
       }
 
       public void map(Object key, Object value, OutputCollector output, Reporter reporter) throws IOException {
              Text t = (Text) value;
              String [] fields = t.toString().split("\\t");
              String address = joinData.get(fields[2]);
              if (address != null) {
                     output.collect(new Text(fields[2]), new Text(fields[0] + "\t" + fields[1] + "\t" + address));
              }
       }
}

 

 

提交任务的类可以这么写：

 

DistributedCache.addCacheFile(new Path(“/address.txt”).toUri(), conf);
FileInputFormat.addInputPath(conf, new Path(“/person.txt”));
FileOutputFormat.setOutputPath(conf, new Path(“/join_out”));
conf.setMapperClass(MapClass.class);
conf.setNumReduceTasks(0);
conf.setOutputKeyClass(Text.class);
conf.setOutputValueClass(TextOutputFormat.class);
JobClient.runJob(conf);

 

 

 

2. join的最小的文件不能完全放进内存，仍然要实现map-side的join

题外话：数据库中，对这种情况的处理是使用grace hash join的算法。例如：SQL Server分别将build input和probe input切分成多个分区部分(partition)，每个partition都包括一个独立的、成对匹配的build input和probe input，这样就将一个大的hash join切分成多个独立、互相不影响的hash join，每一个分区的hash join都能够在内存中完成。SQL Server将切分后的partition文件保存在磁盘上，每次装载一个分区的build input和probe input到内存中，进行一次hash join。

回来正题：

第一步的思考：最小的文件整个数据放不进内存中，那么是否可以考虑可以满足join条件的数据能否放进内存，从而实现In-Memory Hash Join的方式，实现方式很简单：选取小文件（例如名称为small_file），将其参与join的key的数据抽取出来，保存到文件small_file_temp中，small_file_temp这里保证很小，可以完全放到内存中，因此后续的工作时使用small_file_temp文件和大文件进行In-Memory Hash Join。

进一步的思考：

其实保存数据的时候，可以先分区，就是可以根据数据的特征，先将数据划分为多个文件进行保存，实际处理时，只处理其中的一部分数据，这样预先就减少了处理的数据量。

 

如果已经分区了，但最终的小文件是还是放不进内存的，那么还可以怎么做？

方式一：特定的场景下，可以使用bloomfilter。

条件1：小表对于大表中的数据而言，仅仅用于判断大表的数据是否满足join条件，不再从小表中获取其他数据

条件2：不要求数据100%准确，因为bloomfilter存在一定的误判。

可以看到，方式一的最终目的仍然是将操作转换为In-Memory Hash Join的方式。

 

方式二：能否参考数据库中的grace hash join的方式？答案是肯定的。实际的做法是，在保存数据的时候，在分区的基础上（partition），再分桶（bucket），小文件和大文件都拆成很多bucket，小文件和大文件之间的桶的个数只要相互之间是倍数关系即可（是倍数关系，即可相互映射），一般情况下，小文件的桶个数是大文件的2的倍数。这里以hive的实现进行说明：在提交任务之前，先在本地执行一个任务，生成每个大表的桶文件对应的小表的桶文件的hashtable，之后将这些hashtable文件分发到map节点，map节点在执行任务的，根据输入的大表的桶文件，加载进来相应的hashtable文件，之后执行join操作。

注意：分发数据本身就存在一定的代价的，因此若分发数据消耗的时间较多，也许此时reduce-side join的会更快。

 

3. sort merge join的实现

 

上述所讲的操作均没有提及数据是否有序，若数据有序的情况下，完全可以执行类似merge的操作，一边读取数据，一边做join操作。此种join方式不再进一步的探讨。