《数据库实验室MapReduce连接资料讲解》由会员分享，可在线阅读，更多相关《数据库实验室MapReduce连接资料讲解（45页珍藏版）》请在金锄头文库上搜索。
1、厦门大学数据库实验室 MapReduce 连接,连接简介,MapReduce连接策略,连接,连接是关系运算，可以用于合并关系。,在数据库中，一般是表连接操作； 在MapReduce中，连接可以用于合并两个或多个数据集。,例如，用户基本信息和用户活动详情。用户基本信息来自于OLTP数据库。用户活动详情来自于日志文件。,连接的类型,最常用的两个连接类型是内连接和外连接。,内连接比较两个关系中所有的数组，然后生成一个满足连接条件的结果集。,外连接,外连接并不需要两个关系的数组都满足连接条件。在连接条件不满足的时候，外连接可以将一方的数据保留在结果集中。,内连接,左外连接 右外连接 全连接,连接实例,
2、连接简介,MapReduce连接策略,连接,连接是关系运算，可以用于合并关系。,在数据库中，一般是表连接操作； 在MapReduce中，连接可以用于合并两个或多个数据集。,例如，用户基本信息和用户活动详情。用户基本信息来自于OLTP数据库。用户活动详情来自于日志文件。,MapReduce的连接,welcome to use these PowerPoint templates, New Content design, 10 years experience,用户的人口统计信息的聚合操作（例如：青少年和中年人的习惯差异） 当用户超过一定时间没有使用网站后，发邮件提醒他们。 分析用户的浏览习惯，让
3、系统可以提示用户有哪些网站特性还没有使用到，形成一个反馈循环。,MapReduce中的连接策略,重分区连接,复制连接,半连接,reduce端连接。 使用场景：连接两个或多个大型数据集。,map端连接。 使用场景：待连接的数据集中有一个数据集小到可以完全放在缓存中。,map端连接。 使用场景：待连接的数据集中有一个数据集非常大，但同时这个数据集可以被过滤成小到可以放在缓存中。,重分区连接,重分区连接利用MapReduce的排序-合并机制来分组数据。它被实现为使用一个单独的MapReduce任务，并支持多路连接（这里的多路指的是多个数据集）。,Map阶段负责从多个数据集中读取数据，决定每个数据的连
4、接值，将连接值作为输出键。输出值则包含将在reduce阶段被合并的值。,Reduce阶段，一个reducer接收map函数传来的一个输出键的所有输出值，并将数据分为多个分区。在此之后，reducer对所有的分区进行笛卡尔积连接运算，并生成全部的结果集。,在如下示例中，用户数据中有用户姓名，年龄和所在州 $ cat test-data/ch4/users.txtanne 22 NYjoe 39 COalison 35 NYmike 69 VAmarie 27 ORjim 21 ORbob 71 CAmary 53 NYdave 36 VAdude 50 CA 用户活动日志中有用户姓名，进行的动作
5、， 来源IP。这个文件一般都要比用户数据要大得多。 $ cat test-data/ch4/user-logs.txtjim logout 93.24.237.12mike new_tweet 87.124.79.252bob new_tweet 58.133.120.100mike logout 55.237.104.36jim new_tweet 93.24.237.12marie view_user 122.158.130.90,$ hadoop fs -put test-data/ch4/user-logs.txt user-logs.txt $ bin/run.sh com.mann
6、ing.hip.ch4.joins.improved.SampleMain users.txt,user-logs.txt output $ hadoop fs -cat output/part*bob 71 CA new_tweet 58.133.120.100jim 21 OR logout 93.24.237.12jim 21 OR new_tweet 93.24.237.12jim 21 OR login 198.184.237.49marie 27 OR login 58.133.120.100marie 27 OR view_user 122.158.130.90mike 69 V
7、A new_tweet 87.124.79.252mike 69 VA logout 55.237.104.36,重分区连接,过滤（ filter）指的是将map极端的输入数据中不需要的部分丢弃。投影（ project）是关系代数的概念。投影用于减少发送给reducer的字段。,优化重分区连接,传统重分区方法的实现空间效率低下。它需要将连接的所有的输出值都读取到内存中，然后进行多路连接。事实上，如果仅仅将小数据集读取到内存中，然后用小数据集来遍历大数据集，进行连接，这样将更加高效。下图是优化后的重分区连接的流程图。,Map输出的组合键和组合值,上图说明了map输出的组合键和组合值。二次排序将会
8、根据连接键（join key）进行分区，但会用整个组合键来进行排序。组合键包括一个标识源数据集（较大或较小）的整形值，因此可以根据这个整形值来保证较小源数据集的值先于较大源数据的值被reducer接收。,优化重分区连接,上图是实现的类图。类图中包含两个部分，一个通用框架和一些类的实现样例。使用这个连接框架需要实现抽象类 OptimizedDataJoinMapperBase 和 OptimizedDataJoinReducerBase。,OptimizedDataJoinMapperBase,protected abstract Text generateInputTag(String inp
9、utFile); protected abstract boolean isInputSmaller(String inputFile); public void configure(JobConf job) this.inputFile = job.get(map.input.file); this.inputTag = generateInputTag(this.inputFile); if(isInputSmaller(this.inputFile) smaller = new BooleanWritable(true); outputKey.setOrder(0); else smal
10、ler = new BooleanWritable(false); outputKey.setOrder(1); ,这个类的作用是辨认出较小的数据集，并生成输出键和输出值。Configure方法在mapper创建期调用。Configure方法的作用之一是标识每一个数据集，让reducer可以区分数据的源数据集。另一个作用是辨认当前的输入数据是否是较小的数据集。,OptimizedDataJoinMapperBase（续）,protected abstract OptimizedTaggedMapOutput generateTaggedMapOutput(Object value); prot
11、ected abstract String generateGroupKey(Object key, OptimizedTaggedMapOutput aRecord); public void map(Object key, Object value, OutputCollector output, Reporter reporter) throws IOException OptimizedTaggedMapOutput aRecord = generateTaggedMapOutput(value); if (aRecord = null) return; aRecord.setSmal
12、ler(smaller); String groupKey = generateGroupKey(aRecord); if (groupKey = null) return; outputKey.setKey(groupKey); output.collect(outputKey, aRecord);,Map方法首先调用自定义的方法 (generateTaggedMapOutput) 来生成OutputValue对象。这个对象包含了在连接中需要使用的值，和一个标识较大或较小数据集的布尔值。如果map方法可以调用自定义的方法 (generateGroupKey) 来得到可以在连接中使用的键，那么
13、这个键就作为map的输出键。,OptimizedDataJoinReducerBase,public void reduce(Object key, Iterator values, OutputCollector output, Reporter reporter) throws IOException CompositeKey k = (CompositeKey) key; List smaller = new ArrayList(); while (values.hasNext() Object value = values.next(); OptimizedTaggedMapOutpu
14、t cloned =(OptimizedTaggedMapOutput) value).clone(job); if (cloned.isSmaller().get() smaller.add(cloned); else joinAndCollect(k, smaller, cloned, output, reporter); ,Map端处理后已经可以保证较小源数据集的值将会先于较大源数据集的值被接收。这里就可以将所有的较小源数据集的值放到缓存中。在开始接收较大源数据集的值的时候，就开始和缓存中的值做连接操作。,OptimizedDataJoinRuducerBase（续）,protected
15、 abstract OptimizedTaggedMapOutput combine(String key, OptimizedTaggedMapOutput value1,OptimizedTaggedMapOutput value2); private void joinAndCollect(CompositeKey key,List smaller, OptimizedTaggedMapOutput value, OutputCollector output, Reporter reporter) throws IOException if(smaller.size() 1) Optim
16、izedTaggedMapOutput combined = combine(key.getKey(), null, value); collect(key, combined, output, reporter); else for (OptimizedTaggedMapOutput small : smaller) OptimizedTaggedMapOutput combined = combine(key.getKey(), small, value); collect(key, combined, output, reporter); ,方法 joinAndCollect 包含了两个数据集的值，并输出它们。,优化重分区连接实例,例如，需要连接用户详情数据和用户活动日志。,第一步，判断两个数据集中哪一个比较小。对于一般的网站来说，用户详情数据会比较小，用户活动日志会比较大。首先，实现抽象类 OptimizedDataJoinMapperBase。这个将在map端被调用。这个类将创建map的输出键和输出
