Hadoop Mapreduce 连接（Join）之三：半连接（Semi-join）

4rrr

4.1.3 半连接（Semi-join）

假设一个场景，需要连接两个很大的数据集，例如用户日志和OLTP的用户数据。任何一个数据集都不是足够小到可以缓存在map作业的内存中。这样看来，你似乎就不能使用reduce端的连接了。尽管不是必须，你可以问问你自己以下问题：如果一个数据集中有的记录因为无法连接到另一个数据集的记录，将会在连接中被移除。这样还需要将整个数据集放到内存中吗？在这个例子中，在用户日志中的用户仅仅是OLTP用户数据中的用户中的很小的一部分。那么就可以从OLTP用户数据中只取出存在于用户日志中的那部分用户的用户数据。这样就可以得到足够小到可以放在内存中的数据集。这样的解决方案就叫做半连接。

图4.6说明了在半连接中将要执行的三个MapReduce作业（Job）。

让我们看看如何实现一个半连接。

技术20 实现一个半连接

当面对连接两个都很大的数据集的挑战时，很容易想到要用重分区连接，也就是利用了整个MapReduce框架的reduce端的连接。如果你这么想了，又不能够将其中一个数据集过滤到一个较小的尺寸以便放到map端的内存中，这将只是你一个人的想法而已。然而，如果你认为你能够将一个数据集减小到一个可管理的大小，也许就用不着使用重分区连接了。

问题：

需要连接两个都很大的数据集，同时减少整理和排序阶段的消耗。

解决方案：

在这个技术中，将会用到三个MapReduce作业来连接两个数据集，以此来减少reduce端连接的消耗。对于很大的数据集，这个技术非常有用。

讨论：

在这个技术中，将会用到附录D中的复制连接（Replicated join）的代码来实现MapReduce作业中的最后两步。同时，在图4.6中的三个作业将会被分开来说明。

作业1：

第一个MapReduce作业的功能是从日志文件中提取出用户名，用这些用户名生成一个用户名唯一的集合（Set）。这通过让map函数执行了用户名的投影（projection）操作来实现。然后用reducer输出用户名。为了减少在map阶段和reduce阶段之间传输的数据量，就在map任务中采用哈希集（HashSet）来保存用户名，在cleanup方法中输出哈希集的值。图4.7说明了这个作业的流程。

这个MapReduce作业中的代码如下：

public static class Map extends Mapper<Text, Text, Text, NullWritable> {

    private Set<String> keys = new HashSet<String>();

    @Override
    protected void map(Text key, Text value, Context context)
        throws IOException, InterruptedException {
        keys.add(key.toString());
    }

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

        Text outputKey = new Text();

        for(String key: keys) {
            outputKey.set(key);
            context.write(outputKey, NullWritable.get());
        }

    }

}

public static class Reduce extends Reducer<Text, NullWritable, Text, NullWritable> {

    @Override
    protected void reduce(Text key, Iterable<NullWritable> values, Context context)
        throws IOException, InterruptedException {
        context.write(key, NullWritable.get());
    }

}

第一个作业的结果就是来自于日志文件中的所有用户的集合。集合中的用户名是唯一的。

作业2：

第二步是一个复杂的过滤的MapReduce作业。目标是从全体用户的用户数据集中移除不存在于日志文件中的用户。这是一个只包含map的作业。它用到了复制连接来缓存出现在日志文件中的用户名，并把他们和全体用户的数据集连接。来自于作业1的用户唯一的数据集要远远小于全体用户的数据集。很自然的就把小的那个数据集放到缓存中了。图4.8说明了这个作业的流程。

如果你还对附录D中的复制连接框架还不熟悉，那现在快速浏览一下刚刚好。这个框架对 KeyValueTextInputFormat 和 TextOutputFormat 提供了内置支持，并假设 KeyValueTextInputFormat 生成的键是连接键。如此，这也就是数据被展开的过程。图4.9时这个框架的类图。

GenericReplicatedJoin 类是执行连接的类。如图4.9中所示，在 GenericReplicatedJoin 的类列表中前三个类是可扩展的，相对应的复制链接的行为也是可定制的。readFromInputFormat 方法可以用于任意的输入类型（InputFormat）。getDistributedCacheReader 方法可以被重载来支持来自于分布式缓存（distributed cache）的任意文件类型。在这一步中的核心是join方法。join方法将会生成作业的输出键和输出值。在默认的实现中，两个数据集的值将会被合并以生成最终的输出值。你可以改变这个行为，变成值输入来自于用户表的数据，如下所示：

public class ReplicatedFilterJob extends GenericReplicatedJoin {

    @Override
    public Pair join(Pair inputSplitPair, Pair distCachePair) {
        return inputSplitPair;
    }

}

你还需要把来自于作业1的文件放到分布式缓存中：

for(FileStatus f: fs.listStatus(uniqueUserStatus)) {
2     if(f.getPath().getName().startsWith("part")) {
3         DistributedCache.addCacheFile(f.getPath().toUri(), conf);
4     }
5 }

然后，在驱动（driver）代码中，调用 GenericReplicatedJoin 类：

public class ReplicatedFilterJob extends GenericReplicatedJoin {

    public static void runJob(Path usersPath,
                                Path uniqueUsersPath,
                                Path outputPath)
        throws Exception {

        Configuration conf = new Configuration();

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

        Job job = new Job(conf);
        job.setJarByClass(ReplicatedFilterJob.class);
        job.setMapperClass(ReplicatedFilterJob.class);
        job.setNumReduceTasks(0);
        job.setInputFormatClass(KeyValueTextInputFormat.class);
        outputPath.getFileSystem(conf).delete(outputPath, true);
        FileInputFormat.setInputPaths(job, usersPath);
        FileOutputFormat.setOutputPath(job, outputPath);

        if(!job.waitForCompletion(true)) {
            throw new Exception("Job failed");
        }

    }

    @Override
    public Pair join(Pair inputSplitPair, Pair distCachePair) {
        return inputSplitPair;
    }

}

作业2的输出就是已被用户日志数据集的用户过滤过的用户集了。

作业3：

在最后一步中，需要将作业2生成的已过滤的用户集和原始的用户日志合并了。表面上，已过滤的用户集是足够小到可以放到内存中，同样也可以放到分布式缓存中。图4.10说明了这个作业的流程。

FileStatus usersStatus = fs.getFileStatus(usersPath);

for(FileStatus f: fs.listStatus(usersPath)) {

    if(f.getPath().getName().startsWith("part")) {
        DistributedCache.addCacheFile(f.getPath().toUri(), conf);
    }

...

这里要再次用到复制连接框架来执行连接。但这次不用调节join方法的行为，因为两个数据集中的数据都要出现在最后的输出中。

执行这个代码，观察前述步骤生成的输出。

$ bin/run.sh com.manning.hip.ch4.joins.semijoin.Main users.txt user-logs.txt output

$ hadoop fs -ls output
/user/aholmes/output/filtered
/user/aholmes/output/result
/user/aholmes/output/unique

$ hadoop fs -cat output/unique/part*
bob
jim
marie
mike

$ hadoop fs -cat output/filtered/part*
mike 69 VA
marie 27 OR
jim 21 OR
bob 71 CA

$ hadoop fs -cat output/result/part*
jim logout 93.24.237.12 21 OR
mike new_tweet 87.124.79.252 69 VA
bob new_tweet 58.133.120.100 71 CA
mike logout 55.237.104.36 69 VA
jim new_tweet 93.24.237.12 21 OR
marie view_user 122.158.130.90 27 OR
jim login 198.184.237.49 21 OR
marie login 58.133.120.100 27 OR

这些输出说明了在半连接的作业的逻辑进程和最终连接的输出。

小结：

在这个技术中说明了如何使用半连接来合并两个数据集。半连接的构建包括了比其他连接类型更多的步骤。但它确实是一个处理大的数据集的map端连接的强大的工具。当然，这些很大的数据集要能够被减小到能够放到内存中。

附录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端的连接。但是它需要有序的待连接的数据集的输入文件，并要求将其分发到相同的分区中。这样就造成了繁重的预处理工作。