MapReduce Design Patterns（5.表连接）（九） Chapter 5. Join Patterns

http://blog.csdn.net/cuirong1986/article/details/8485273

把数据保存成一个巨大的数据集不是很常见。例如，用户信息数据频繁更新，所以要保存到关系数据库中。于此同时，web日志以恒定的数据流量增加，直接写到HDFS。这些日志的日常分析过的数据保存在hdfs的某个地方，财务数据存储在加密的仓库中。还有很多例子。。。

（原文are stored someone where in HDFS 貌似应改为 are stored somewhere in HDFS）

 

数据遍布于各处，本身也很有价值。当我们合起来分析这些数据集时会发现一些有趣的关系。这就是join模式可以使用的地方。Join可以用一个小的引用集合让数据更丰富，或者过滤出或选择出指定的一系列类型。这种使用也有很多。

 

在RDB中，join操作可以使用简单的命令完成，数据库引擎会处理所有工作。对我们不幸的是，MapReduce中的join不会这么简单。MapReduce每次操作一个键值对，一般来自相同的输入。我们现在会处理至少两个输入数据集而且很可能有不同的结构，所以我们需要知道记录来自哪个数据集，以便正确的处理。一般情况下，join操作之前不会过滤数据，所以一些join操作需要把每个输入的字节都发送到reduce阶段，网络传输较繁重。例如，拿一个1T的数据跟另一个1T的数据做join，至少需要2T的网络流量-而且是在实际的join逻辑之前做的。

 

基于所有以上复杂的原因，要从几个不同的方式中选出一个最好的方式。由于框架只是简单的分解成map和reduce任务，有很多工作需要手动处理，有很多事情要考虑。当你了解了这些可能性，问题就是什么时候用什么模式。不管用什么MapReduce操作，网络流量都是非常重要的资源，join更会大量的使用它。让网络传输更有效率是值得考虑的，网络优化是这些模式中不同的地方。

 

下面讲到的每种模式都可以使用其执行inner join或至少一种类型outer join。至于选那个模式，很大程度上取决于数据集的大小，格式，哪种类型的join。另一方面，笛卡尔积也完全不同，但当我们遇到时，总能想办法解决。

 

本章第一个模式是reduce side join，最基本的，同时使用bloom filter的改进版本。随后，我们讨论两种map端执行join操作的模式：使用分布式缓存或hadoop MapReduce api里的归并特性。最终，我们会看到怎样巧妙的执行笛卡尔积。

 

对你的情况选择正确类型的join比较困难。注意下面“Applicability”部分的描述。

A Refresher on Joins

如果你有深厚的RDB sql背景出身，可以跳过这一部分，对于hadoop初学者来说，join可能有点像外部的东西。

Join很可能是你执行的MapReduce中最复杂的操作之一（我觉得没有之一）。在设计上MapReduce依靠查看每一条记录或隔离分组，非常适合处理巨大数据集，所以join两个非常大的数据集可能不太符合其范例。在分析模式之前，让我们先搞清楚什么是join，还有不同的join类型。

 

Join是一种基于一个或几个字段，连接两个或两个以上数据集的一种操作，例如所谓的外键。外键是一种关系型表中匹配另一张表的某列的字段。并且是表示表与表之间相互关系的重要手段。用例子说明是最简单的方式了，下面深入研究。

 

为了简单的解释join类型，使用两个数据集A和B。外键定义为f。不同的join类型分别如表5-1和5-2表示。随后的描述也会用到这两张表

Table 5-1. Table A

Table 5-2. Table B

INNER JOIN

当人们不指定类型，说join时，通常指的是inner join。用这种类型，A表和B表具有相同外键值得记录才被取出来，如果把这个结果放在一张新表里，结果中在A表不在B表，或在B表不在A表的数据都不会出现在新表里。

表5-3 展示了这个结果集。使用userid作为外键。

Table 5-3. Inner Join of A + B on User ID

用户id是3或5的在两个表里都有，所以会出现在最终的表里。用户id是4或9在表A，用户id是8的在表Ｂ，但都不在另一个表中，所以被忽略。然而这些记录会出现在下面介绍的类型中join类型中。

 

OUTER JOIN

外连接跟内连接相似，但只出现在一个表的记录会出现在最终结果表。有三种外连接，他们直接决定哪些没有匹配的记录会出现在最终表里。

 

左外连接中，左边表里没匹配的记录会出现在最终表里，右边表没有与之对应的记录位置填null值，右外连接相反。全外连接会包含两个表所有的数据，像是一种左外连接和右外连接的结合。

表5-4展示了左外连接结果。

Table 5-4. Left Outer Join of A + B on User ID

用户id为3或5的用户两个表都有，所以结果有。4和9在A表有，并且A表是左表，所以这两个会保留，右表字段位置填null。

 

表5-5是右外连接结果。

Table 5-5. Right Outer Join of A + B on User ID

同理，3,5记录两表都有，所以结果有。作为右外连接，8出现在右表，左边表的响应位置填null。

 

表5-6是全外连接的结果。

Table 5-6. Full Outer Join of A + B on User ID

不管匹配的不匹配的记录都出现在结果中。

 

ANTIJOIN

反连接是全外连接的结果减掉内连接后的结果。就是说最终的结果不包含满足外键关联的记录。

表5-7展示了反连接。

Table 5-7. Antijoin of A + B on User ID

 

CARTESIAN PRODUCT

笛卡尔积或交叉乘积指用一张表的每条记录匹配另一张表的每条记录。如果表X和表Y分别有n，m条记录，则X表和Y表的笛卡尔积，计做X × Y,包含n*m条记录。与其他join不同，笛卡尔积没有外键的概念。这种操作无论怎么实现，代价都很高，MapReduce也不例外。

 

表5-8展示了笛卡尔积。

Table 5-8. Cartesian Product, A × B

 

Reduce Side Join

Pattern Description

Reduce端join与其他join模式相比执行时间是最长的，不过实现起来比其它的简单。

Intent

靠某些外键join多个大数据集。

Motivation

Reduce端的join被证明是MapReduce join中最简单的实现，因此是个不错的选择。它能执行上面说的所有join，并相对容易，也没有对数据集的限制。能做同样多的数据集之间的join。

Reduce端join需要大量的网络传输，因为大量数据发送到reduce。如果你的资源允许并不太关注执行时间，就可以用它。但是，如果要join的数据非常巨大，就只能选这种join。

Applicability

可以在一下情况下用：

·多个大的数据集用外键做join。如果除了一个之外，其它数据集都可放进内存，从哪个是用复制join。

·你想得到能执行任意join操作的灵活性。

Structure

·mapper准备要join的数据，并从记录中抽取外键，当做输出key，整条记录作为输出值。输出值要标记上属于哪个数据集，

·可以使用hash partitioner，或自定义，让数据分发的更均匀。

·reducer收集每个输入组的值到临时列表，执行期望的join操作。例如标记为A的记录存储在代表A的列表，标记为B的记录存储在代表B的列表。然后迭代两个集合的所有记录join到一起。对于内连接，两个列表都不是空的就输出join后的数据。对于外连接，空列表也会跟非空列表join。反连接会确保只有一个列表是空的。非空列表记录跟空列表一块写出来。

Figure 5-1. The structure of the reduce side join pattern

Consequences

输出是与reduce任务对应的一定数量的文件。每一个部分文件包含join过的一部分记录。Join结果包含的列由如何join决定。如果是外连接或反连接，会出现null值。

Resemblances

Sql：join在sql中很容易执行：

SELECT users.ID, users.Location, comments.upVotes

FROM users

[INNER|LEFT|RIGHT] JOIN comments

ON users.ID=comments.UserID

 

Pig：pig支持内外连接。

-- Inner Join

A = JOIN comments BY userID, users BY userID;

-- Outer Join

A = JOIN comments BY userID [LEFT|RIGHT|FULL] OUTER, users BY userID;

 

Performance analysis

一个简单的reduce端join会给集群网络带来负担。因为每条记录的外键抽取出来并跟整条记录一同输出，此时没有数据被过滤掉，非常多的数据会发送到混洗和排序阶段。基于此原因，reduce端join应该相对于普通分析使用较多的reducer。

如果本章其它模式能使用（笛卡尔积除外），就用那个模式。有时这种基本的join模式是具体环境下唯一可用的方法。

Reduce Side Join Example

User and comment join

在这个例子中，使用*的用户表和评论表。分表存储数据是在情理之中的，因为对每一条评论都存储一次用户信息是没必要的。这将使更新用户信息比较困难。然而，当需要关联一条评论和用户时，分开的数据集会带来不便。通过reduce端 join的使用，可以通过userid作为外键join在一起。这个例子中，我们根据配置选择执行内，外，反连接。

 

Hadoop支持多输入数据类型，可以对不同数据源的输入分片创建不同的Mapper类和输入格式。这非常有用，因为你不必在同一个map中区分两个不同输入数据的逻辑了。下面的例子中，创建了两个mapper类：一个是用户数据，一个是评论数据。；每个mapper类都输出userid作为外键，整条记录和一个标志所属数据集的字符作为输出值。然后reducer对每个输入组拷贝数据到内存，并记录数据来源。然后做join操作并输出。

Notice：建议所有mapper的输出键和值得类型相同。

 

问题：给出用户数据和用户评论数据，把评论所属用户信息关联到评论数据上。

 

Driver code。由于使用多输入，job 配置跟标准配置稍有不同。也通过job配置join类型为命令行第二个参数，方便在reducer中使用。相关代码如下：

 

// Use MultipleInputs to set which input uses what mapper
// This will keep parsing of each data set separate from a logical standpoint
// The first two elements of the args array are the two inputs
MultipleInputs.addInputPath(job, new Path(args[0]), TextInputFormat.class,UserJoinMapper.class);
MultipleInputs.addInputPath(job, new Path(args[1]),TextInputFormat.class,CommentJoinMapper.class);
job.getConfiguration()..set("join.type", args[2]);

 

user mapper code。Mapper解析用户数据，抽取userid作为key，记录前加字符A作为value，输出。

 

public static class UserJoinMapper extends Mapper<Object, Text, Text, Text> {

    private Text outkey = new Text();
    private Text outvalue = new Text();

    public void map(Object key, Text value, Context context)
            throws IOException, InterruptedException {
// Parse the input string into a nice map
        Map<String, String> parsed = MRDPUtils.transformXmlToMap(value.toString());
        String userId = parsed.get("Id");
// The foreign join key is the user ID
        outkey.set(userId);
// Flag this record for the reducer and then output
        outvalue.set("A" + value.toString());
        context.write(outkey, outvalue);
    }
}

Notice：从map端输出值得时候，不必将整个记录发送，也可以只发送你需要的字段。这需要在map端进行更多的处理，但是这是值得的。另外，外键作为输出key，不必把它放在value里。

 

Comment mapper code。解析输入数据，跟joinMapper的使用类似，同样抽取用户id，记录前加字符B作为输出值。

 

public static class CommentJoinMapper extends
        Mapper<Object, Text, Text, Text> {

    private Text outkey = new Text();
    private Text outvalue = new Text();

    public void map(Object key, Text value, Context context)
            throws IOException, InterruptedException {
        Map<String, String> parsed = transformXmlToMap(value.toString());
// The foreign join key is the user ID
        outkey.set(parsed.get("UserId"));
// Flag this record for the reducer and then output
        outvalue.set("B" + value.toString());
        context.write(outkey, outvalue);
    }
}

 

Reducer code。迭代所有值，对每个输入组，通过查看记录的标记放到其中一个列表中。完成上面这一步之后，就用这两个列表完成join操作。Join逻辑根据join类型的不同稍有不同，但总会涉及到迭代两个列表并写到上context对象。Join的类型在setup阶段从job配置中获得。Reduce方法的主要代码：

 

 

public static class UserJoinReducer extends Reducer<Text, Text, Text, Text> {

    private static final Text EMPTY_TEXT = new Text("");
    private Text tmp = new Text();
    private ArrayList<Text> listA = new ArrayList<Text>();
    private ArrayList<Text> listB = new ArrayList<Text>();
    private String joinType = null;

    public void setup(Context context) {
// Get the type of join from our configuration
        joinType = context.getConfiguration().get("join.type");
    }

    public void reduce(Text key, Iterable<Text> values, Context context)
            throws IOException, InterruptedException {
// Clear our lists
        listA.clear();
        listB.clear();
// iterate through all our values, binning each record based on what
// it was tagged with. Make sure to remove the tag!
        while (values.hasNext()) {
            tmp = values.next();
            if (tmp.charAt(0) == 'A') {
                listA.add(new Text(tmp.toString().substring(1)));
            } else if (tmp.charAt('0') == 'B') {
                listB.add(new Text(tmp.toString().substring(1)));
            }
        }
// Execute our join logic now that the lists are filled
        executeJoinLogic(context);
    }

    private void executeJoinLogic(Context context) throws IOException,
            InterruptedException {
    }
}

Reducer的输入类型是两个Text 。输入key是外键连接key，本例中是用户id。跟外键关联的输入值的记录从用户表来的标记为‘A’（原文有误），从评论表来的标记为‘B‘。你想执行的任何类型的数据结构应在输出之前准备好。简单起见，原来来自左边数据集（用户）的xml值作为输出key，右边的作为value。

 

下面，来看每种join的代码。首先是内连接。两个列表都不是空的，执行嵌套循环join。

 

 

        if (joinType.equalsIgnoreCase("inner")) {
// If both lists are not empty, join A with B
            if (!listA.isEmpty() && !listB.isEmpty()) {
                for (Text A : listA) {
                    for (Text B : listB) {
                        context.write(A, B);
                    }
                }
            }
        }
//左外连接，右边列表是空的，输出空值：
        else if(joinType.equalsIgnoreCase("leftouter")) {
// For each entry in A,
            for (Text A : listA) {
// If list B is not empty, join A and B
                if (!listB.isEmpty()) {
                    for (Text B : listB) {
                        context.write(A, B);
                    }
                } else {
// Else, output A by itself
                    context.write(A, EMPTY_TEXT);
                }
            }
        }
//右外连接：
        else if(joinType.equalsIgnoreCase("rightouter")) {
// For each entry in B,
            for (Text B : listB) {
// If list A is not empty, join A and B
                if (!listA.isEmpty()) {
                    for (Text A : listA) {
                        context.write(A, B);
                    }
                } else {
// Else, output B by itself
                    context.write(EMPTY_TEXT, B);
                }
            }
        }
//全外连接：
        else if(joinType.equalsIgnoreCase("fullouter")) {
// If list A is not empty
            if (!listA.isEmpty()) {
// For each entry in A
                for (Text A : listA) {
// If list B is not empty, join A with B
                    if (!listB.isEmpty()) {
                        for (Text B : listB) {
                            context.write(A, B);
                        }
                    } else {
// Else, output A by itself
                        context.write(A, EMPTY_TEXT);
                    }
                }
            } else {
// If list A is empty, just output B
                for (Text B : listB) {
                    context.write(EMPTY_TEXT, B);
                }
            }
        }
//反连接：
        else if(joinType.equalsIgnoreCase("anti")) {
// If list A is empty and B is empty or vice versa
            if (listA.isEmpty() ^ listB.isEmpty()) {
// Iterate both A and B with null values
// The previous XOR check will make sure exactly one of
// these lists is empty and therefore the list will be
// skipped
                for (Text A : listA) {
                    context.write(A, EMPTY_TEXT);
                }
                for (Text B : listB) {
                    context.write(EMPTY_TEXT, B);
                }
            }
        }

 

Notice：考虑到随后的数据分析确保用合适的字段分隔符。输出空文本是不明智的。而应该用合适的结构代替。这对接下来的分析有好处。

 

Combiner optimization。因为join逻辑在reduce端完成，combiner不会有太大作用。

Reduce Side Join with Bloom Filter

Reputable user and comment join

本例与前面的不同之处是使用bloom filter对mapper的输出过滤。这会减少发送到reducer的数据量，减少运行时间。比如我们只对某些专家级用户感兴趣。例如，声誉值在1500以上。标准的reduce 端join不能使用，加上了额外的条件—声誉值在1500以上的记录才写到context对象。这需要解析所有的数据。把join阶段不需要的数据过滤掉，就可以减少网络IO。使用bloom filter对内连接特别有用，对全外连接和反连接可能用处不大。因为这两个操作需要把所有数据发送到reducer。

 

过滤出不满足条件的记录用UserJoinMapper 类就可以了，因为用户声誉值就在数据里。但评论数据里没有用户声誉值，并且数据量比用户大。通过bloom filter的使用，用少量内存就能完成我们需要的检测。预处理阶段需要训练bloom filter，用1500声誉值以上的用户。

 

随后的例子，两个mapper都跟前面的稍有不同。UserJoinMapper 类增加了声誉值的检测。CommentJoin 类从分布式缓存反序列化bloom filter，用它在输出之前做检测。Reduce不变。驱动代码稍有不同：用分布式缓存存储bloom filter。关于它的使用此处省略，可以参考附录A。

 

User mapper code。解析出声誉值大于1500的记录，用户id作为key，记录作为value输出。

 

public static class UserJoinMapper extends Mapper<Object, Text, Text, Text> {

    private Text outkey = new Text();
    private Text outvalue = new Text();

    public void map(Object key, Text value, Context context)
            throws IOException, InterruptedException {
        Map<String, String> parsed = transformXmlToMap(value.toString());
// If the reputation is greater than 1,500,
// output the user ID with the value
        if (Integer.parseInt(parsed.get("Reputation")) > 1500) {
            outkey.set(parsed.get("Id"));
            outvalue.set("A" + value.toString());
            context.write(outkey, outvalue);
        }
    }
}

Comment mapper code。Bloom filter反序列化出来，调用map方法检测user id。检测通过，记录就同外键（user id）一同输出。

 

public static class CommentJoinMapperWithBloom extends
        Mapper<Object, Text, Text, Text> {

    private BloomFilter bfilter = new BloomFilter();
    private Text outkey = new Text();
    private Text outvalue = new Text();

    public void setup(Context context) {
        Path[] files;
        try {
            files = DistributedCache.getLocalCacheFiles(context.getConfiguration());
            DataInputStream strm = new DataInputStream(new FileInputStream(new File(files[0].toString())));
            bfilter.readFields(strm);
        } catch (Exception e) {
// TODO Auto-generated catch block
            e.printStackTrace();
        }
    }

    public void map(Object key, Text value, Context context)
            throws IOException, InterruptedException {
        Map<String, String> parsed = transformXmlToMap(value.toString());
        String userId = parsed.get("UserId");
        if (bfilter.membershipTest(new Key(userId.getBytes()))) {
            outkey.set(userId);
            outvalue.set("B" + value.toString());
            context.write(outkey, outvalue);
        }
    }
}

 

Notice：用这种算法，reducer中就不用验证声誉值。如果有false positive的记录从CommentJoinMapperWithBloom输出，也没关系，因为reduce端没有与之join的用户数据。

用户数据只取声誉值大于1500的就能100%保证正确率。使用bloom filter的好处是大大减少了评论数据发送到reduce的数据量。也考虑了bloom filter 的false positives特性可能对程序的影响。