Hadoop Mapreduce 连接（Join）之二：复制连接（Replication join）

4rrr

4.1.2 复制连接（Replication join）

复制连接是map端的连接。复制连接得名于它的具体实现：连接中最小的数据集将会被复制到所有的map主机节点。复制连接有一个假设前提：在被连接的数据集中，有一个数据集足够小到可以缓存在内存中。

如图4.5所示，MapReduce复制连接工作原理如下：

• 我们将使用分布式缓存（Districubted cache）将这个小数据集复制到所有运行map任务的节点。
• 用各个map任务初始化方法将这个小数据集装载到一个哈希表（hashtable）中。
• 逐条用大数据集中的记录遍历这个哈希表，逐个判断是否符合连接条件。
• 输出符合连接条件的结果。
  

复制连接的实现非常直接明了。更具体的内容可以参考《Hadoop in Action》。附录D提供了一个通用的框架来实现复制连接。这个框架支持任意类型的InputFormat和OutputFormat的数据。（我们将在下一个技术中使用这个框架。）复制连接框架根据内存足迹的大小从分布式缓存的内容和输入块（input split）两者中动态地决定需要缓存的对象。

如果所有的输入数据集都不能够小到可以放到缓存中，那有没有办法来优化map端连接呢？那就到了看半连接（semi-join）的时间了。

附录D.2 一个复制连接框架

复制连接是map端连接，得名于它的具体实现：连接中最小的数据集将会被复制到所有的map主机节点。复制连接的实现非常直接明了。更具体的内容可以参考Chunk Lam的《Hadoop in Action》。

这个部分的目标是：创建一个可以支持任意类型的数据集的通用的复制连接框架。这个框架中提供了一个优化的小功能：动态监测分布式缓存内容和输入块的大小，并判断哪个更大。如果输入块较小，那么你就需要将map的输入块放到内存缓冲中，然后在mapper的cleanup方法中执行连接操作了。

图D.4是这个框架的类图，这里我提供了连接类（ GenericReplicatedJoin）的具体实现，而不仅仅是一个抽象类。在这个框架外，这个类将和 KeyValueTextInputFormat 及 TextOutputFormat 协作。这里有一个假设前提：每个数据文件的第一个标记是连接键。此外，连接类也可以被继承扩展来支持任意类型的输入和输出。

图D.5是连接框架的算法。Mapper的setup方法判断在map的输入块和分布式缓存的内容中哪个大。如果分布式缓存的内容比较小，那么它将被装载到内存缓存中。Map函数开始连接操作。如果输入块比较小，map函数将输入块的键\值对装载到内存缓存中。Map的cleanup方法将从分布式缓存中读取记录，逐条记录和在内存缓存中的键\值对进行连接操作。

以下代码的 GenericReplicatedJoin 中的setup方法是在map的初始化阶段调用的。这个方法判断分布式缓存中的文件和输入块哪个大。如果文件比较小，则将文件装载到HashMap中。

@Override
protected void setup(Context context)
    throws IOException, InterruptedException {

    distributedCacheFiles = DistributedCache.getLocalCacheFiles(context.getConfiguration());
    int distCacheSizes = 0;

    for (Path distFile : distributedCacheFiles) {
        File distributedCacheFile = new File(distFile.toString());
        distCacheSizes += distributedCacheFile.length();
    }

    if(context.getInputSplit() instanceof FileSplit) {
        FileSplit split = (FileSplit) context.getInputSplit();
        long inputSplitSize = split.getLength();
        distributedCacheIsSmaller = (distCacheSizes < inputSplitSize);
    } else {
        distributedCacheIsSmaller = true;
    }

    if (distributedCacheIsSmaller) {
        for (Path distFile : distributedCacheFiles) {
            File distributedCacheFile = new File(distFile.toString());
            DistributedCacheFileReader reader = getDistributedCacheReader();
            reader.init(distributedCacheFile);

            for (Pair p : (Iterable<Pair>) reader) {
                addToCache(p);
            }

            reader.close();
        }
    }
}

Map方法将会根据setup方法是否将分布式缓存的内容装载到内存的缓存中来选择行为。如果分布式缓存中的内容被装载到内存中，那么map方法就将输入块的记录和内存中的缓存做连接操作。如果分布式缓存中的内容没有被装载到内存中，那么map方法就将输入块的记录装载到内存中，然后在cleanup方法中使用。

@Override
protected void map(Object key, Object value, Context context)
    throws IOException, InterruptedException {
    Pair pair = readFromInputFormat(key, value);

    if (distributedCacheIsSmaller) {
        joinAndCollect(pair, context);
    } else {
        addToCache(pair);
    }
}

public void joinAndCollect(Pair p, Context context)
    throws IOException, InterruptedException {
    List<Pair> cached = cachedRecords.get(p.getKey());

    if (cached != null) {
        for (Pair cp : cached) {
            Pair result;

            if (distributedCacheIsSmaller) {
                result = join(p, cp);
            } else {
                result = join(cp, p);
            }

            if (result != null) {
                context.write(result.getKey(), result.getData());
            }
        }
    }
}

public Pair join(Pair inputSplitPair, Pair distCachePair) {
    StringBuilder sb = new StringBuilder();

    if (inputSplitPair.getData() != null) {
        sb.append(inputSplitPair.getData());
    }

    sb.append("\t");

    if (distCachePair.getData() != null) {
        sb.append(distCachePair.getData());
    }

    return new Pair<Text, Text>(
                new Text(inputSplitPair.getKey().toString()),
                new Text(sb.toString()));
}

当所有的记录都被传输给map方法后，MapReduce将会调用cleanup方法。如果分布式缓存中的内容比输入块大，连接将会在cleanup中进行。连接的对象是map函数的缓存中的输入块的记录和分布式缓存中的记录。

@Override
protected void cleanup(Context context)
    throws IOException, InterruptedException {

    if (!distributedCacheIsSmaller) {

        for (Path distFile : distributedCacheFiles) {
            File distributedCacheFile = new File(distFile.toString());
            DistributedCacheFileReader reader = getDistributedCacheReader();
            reader.init(distributedCacheFile);

            for (Pair p : (Iterable<Pair>) reader) {
                joinAndCollect(p, context);
            }

            reader.close();
        }
    }
}

最后，作业的驱动代码必须指定需要装载到分布式缓存中的文件。以下的代码可以处理一个文件，也可以处理MapReduce输入结果的一个目录。

Configuration conf = new Configuration();

FileSystem fs = smallFilePath.getFileSystem(conf);
FileStatus smallFilePathStatus = fs.getFileStatus(smallFilePath);

if(smallFilePathStatus.isDir()) {
    for(FileStatus f: fs.listStatus(smallFilePath)) {
        if(f.getPath().getName().startsWith("part")) {
            DistributedCache.addCacheFile(f.getPath().toUri(), conf);
        }
    }
} else {
    DistributedCache.addCacheFile(smallFilePath.toUri(), conf);
}

这个框架假设分布式缓存中的内容和输入块的内容都可以被装载到内存中。这个框架的优点在于它只会将两者之中较小的装载到内存中。

在论文《A Comparison of Join Algorithms for Log Processing in MapReduce》中，你可以看到这个方法对于分布式缓存中的内容较大时的场景的更进一步的优化。在他们的优化中，他们将分布式缓存分成N个分区，并将输入块放入N个哈希表。这样在cleanup方法中的优化就更加高效。

在map端的复制连接的问题在于，map任务必须在启动时读取分布式缓存。上述论文提到的另一个优化方案是重载FileInputFormat的splitting。将存在于同一个主机上的输入块合并成一个块。然后就可以减少需要装载分布式缓存的map任务的个数了。

最后一个说明，Hadoop在org.apache.hadoop.mapred.join包中自带了map端的连接。但是它需要有序的待连接的数据集的输入文件，并要求将其分发到相同的分区中。这样就造成了繁重的预处理工作。