- 浏览: 2148290 次
- 性别:
- 来自: 北京
文章分类
- 全部博客 (682)
- 软件思想 (7)
- Lucene(修真篇) (17)
- Lucene(仙界篇) (20)
- Lucene(神界篇) (11)
- Solr (48)
- Hadoop (77)
- Spark (38)
- Hbase (26)
- Hive (19)
- Pig (25)
- ELK (64)
- Zookeeper (12)
- JAVA (119)
- Linux (59)
- 多线程 (8)
- Nutch (5)
- JAVA EE (21)
- Oracle (7)
- Python (32)
- Xml (5)
- Gson (1)
- Cygwin (1)
- JavaScript (4)
- MySQL (9)
- Lucene/Solr(转) (5)
- 缓存 (2)
- Github/Git (1)
- 开源爬虫 (1)
- Hadoop运维 (7)
- shell命令 (9)
- 生活感悟 (42)
- shell编程 (23)
- Scala (11)
- MongoDB (3)
- docker (2)
- Nodejs (3)
- Neo4j (5)
- storm (3)
- opencv (1)
最新评论
-
qindongliang1922:
粟谷_sugu 写道不太理解“分词字段存储docvalue是没 ...
浅谈Lucene中的DocValues -
粟谷_sugu:
不太理解“分词字段存储docvalue是没有意义的”,这句话, ...
浅谈Lucene中的DocValues -
yin_bp:
高性能elasticsearch ORM开发库使用文档http ...
为什么说Elasticsearch搜索是近实时的? -
hackWang:
请问博主,有用solr做电商的搜索项目?
Solr中Group和Facet的用法 -
章司nana:
遇到的问题同楼上 为什么会返回null
Lucene4.3开发之第八步之渡劫初期(八)
我们都知道在数据库里,多个表之间是可以根据某个链接键进行join的,这也是数据库的范式规范,通过主外键的关联,由此来减少数据冗余,提高性能。当然近几年,随着NOSQL的兴起,出现了基于列的的列式数据库,典型的有Hbase,MongonDB,Cassdran,等等,NOSQL数据库弱化了关联,直接将一整条数据,存入一列,以及去掉了数据库的部分事务特性,从而在海量数据面前显得游刃有余,当然,大部分的NOSQL不支持join操作,也没有绝对的必要支持,因为现在,我们完全是把一整条数据存在了一起,虽然多了许多冗余,但也换来了比较高检索性能,扩展性能,可靠性能。但某些业务场景下,我们仍然需要Join操作,这时候怎么办?
如果数据量比较大的情况下,我们可以使用Hadoop的MapReduce来完成大表join,尤其对Hbase的某些表进行join操作,当然我们也可以使用Hive或Pig来完成,其实质在后台还是运行的一个MR程序。
那么,散仙今天就来看下如何使用MapReduce来完成一个join操作,Hadoop的join分为很多种例如;Reduce链接,Map侧链接,半链接和Reduce侧链接+BloomFilter等等,各个链接都有自己特定的应用场景,没有绝对的谁好谁坏。
今天散仙要说的是,基于Reduce侧的链接,原理如下:
1、在Reudce端进行连接。
在Reudce端进行连接是MapReduce框架进行表之间join操作最为常见的模式,其具体的实现原理如下:
Map端的主要工作:为来自不同表(文件)的key/value对打标签以区别不同来源的记录。然后用连接字段作为key,其余部分和新加的标志作为value,最后进行输出。
reduce端的主要工作:在reduce端以连接字段作为key的分组已经完成,我们只需要在每一个分组当中将那些来源于不同文件的记录(在map阶段已经打标志)分开,最后进行笛卡尔只就ok了。
本次的实现是基于hadoop的旧版API+contribu扩展包里的,DataJoin的工具类辅助来完成的,下篇博客,将会给出,基于新版API,独立来完成Reduce侧的连接示例。
现在看下散仙的两个文件的测试数据,一个是a.txt,另一个是b.txt
源码如下:
运行,日志
运行结果,如下图所示:
可以看出,MR正确的完成了join操作,需要注意的是Reduce侧连接的不足之处:
之所以会存在reduce join这种方式,我们可以很明显的看出原:因为整体数据被分割了,每个map task只处理一部分数据而不能够获取到所有需要的join字段,因此我们需要在讲join key作为reduce端的分组将所有join key相同的记录集中起来进行处理,所以reduce join这种方式就出现了。这种方式的缺点很明显就是会造成map和reduce端也就是shuffle阶段出现大量的数据传输,效率很低。
另外一点需要注意的是,散仙在eclipse里进行调试,Local模式下会报异常,建议提交到hadoop的测试集群上进行测试。
如果数据量比较大的情况下,我们可以使用Hadoop的MapReduce来完成大表join,尤其对Hbase的某些表进行join操作,当然我们也可以使用Hive或Pig来完成,其实质在后台还是运行的一个MR程序。
那么,散仙今天就来看下如何使用MapReduce来完成一个join操作,Hadoop的join分为很多种例如;Reduce链接,Map侧链接,半链接和Reduce侧链接+BloomFilter等等,各个链接都有自己特定的应用场景,没有绝对的谁好谁坏。
今天散仙要说的是,基于Reduce侧的链接,原理如下:
1、在Reudce端进行连接。
在Reudce端进行连接是MapReduce框架进行表之间join操作最为常见的模式,其具体的实现原理如下:
Map端的主要工作:为来自不同表(文件)的key/value对打标签以区别不同来源的记录。然后用连接字段作为key,其余部分和新加的标志作为value,最后进行输出。
reduce端的主要工作:在reduce端以连接字段作为key的分组已经完成,我们只需要在每一个分组当中将那些来源于不同文件的记录(在map阶段已经打标志)分开,最后进行笛卡尔只就ok了。
本次的实现是基于hadoop的旧版API+contribu扩展包里的,DataJoin的工具类辅助来完成的,下篇博客,将会给出,基于新版API,独立来完成Reduce侧的连接示例。
现在看下散仙的两个文件的测试数据,一个是a.txt,另一个是b.txt
a文件的数据 1,三劫散仙,13575468248 2,凤舞九天,18965235874 3,忙忙碌碌,15986854789 4,少林寺方丈,15698745862
b文件的数据 3,A,99,2013-03-05 1,B,89,2013-02-05 2,C,69,2013-03-09 3,D,56,2013-06-07
源码如下:
package com.qin.reducejoin; import java.io.DataInput; import java.io.DataOutput; import java.io.IOException; import java.util.Iterator; import org.apache.hadoop.conf.Configuration; import org.apache.hadoop.conf.Configured; import org.apache.hadoop.fs.FileSystem; import org.apache.hadoop.fs.Path; import org.apache.hadoop.io.Text; import org.apache.hadoop.io.Writable; import org.apache.hadoop.mapred.FileInputFormat; import org.apache.hadoop.mapred.FileOutputFormat; import org.apache.hadoop.mapred.JobClient; import org.apache.hadoop.mapred.JobConf; import org.apache.hadoop.mapred.KeyValueTextInputFormat; import org.apache.hadoop.mapred.MapReduceBase; import org.apache.hadoop.mapred.Mapper; import org.apache.hadoop.mapred.OutputCollector; import org.apache.hadoop.mapred.Reducer; import org.apache.hadoop.mapred.Reporter; import org.apache.hadoop.mapred.TextInputFormat; import org.apache.hadoop.mapred.TextOutputFormat; import org.apache.hadoop.util.ReflectionUtils; import org.apache.hadoop.util.Tool; import org.apache.hadoop.util.ToolRunner; import org.apache.hadoop.contrib.utils.join.DataJoinMapperBase; import org.apache.hadoop.contrib.utils.join.DataJoinReducerBase; import org.apache.hadoop.contrib.utils.join.TaggedMapOutput; import com.qin.joinreduceerror.JoinReduce; /*** * * Hadoop1.2的版本,旧版本实现的Reduce侧连接 * * @author qindongliang * * 大数据交流群:376932160 * 搜索技术交流群:324714439 * * */ public class DataJoin extends Configured implements Tool { /** * * Map实现 * * */ public static class MapClass extends DataJoinMapperBase { /** * 读取输入的文件路径 * * **/ protected Text generateInputTag(String inputFile) { //返回文件路径,做标记 return new Text(inputFile); } /*** * 分组的Key * * **/ protected Text generateGroupKey(TaggedMapOutput aRecord) { String line = ((Text) aRecord.getData()).toString(); String[] tokens = line.split(","); String groupKey = tokens[0]; return new Text(groupKey); } protected TaggedMapOutput generateTaggedMapOutput(Object value) { TaggedWritable retv = new TaggedWritable((Text) value); retv.setTag(this.inputTag); return retv; } } /** * * Reduce进行笛卡尔积 * * **/ public static class Reduce extends DataJoinReducerBase { /*** * 笛卡尔积 * * */ protected TaggedMapOutput combine(Object[] tags, Object[] values) { if (tags.length < 2) return null; String joinedStr = ""; for (int i=0; i<values.length; i++) { if (i > 0) {joinedStr += ",";} TaggedWritable tw = (TaggedWritable) values[i]; String line = ((Text) tw.getData()).toString(); String[] tokens = line.split(",", 2); joinedStr += tokens[1]; } TaggedWritable retv = new TaggedWritable(new Text(joinedStr)); retv.setTag((Text) tags[0]); return retv; } } /** * * 自定义的输出类型 * * ***/ public static class TaggedWritable extends TaggedMapOutput { private Writable data; /** * 注意加上构造方法 * * */ public TaggedWritable() { // TODO Auto-generated constructor stub } public TaggedWritable(Writable data) { this.tag = new Text(""); this.data = data; } public Writable getData() { return data; } public void write(DataOutput out) throws IOException { this.tag.write(out); //此行代码很重要 out.writeUTF(this.data.getClass().getName()); this.data.write(out); } public void readFields(DataInput in) throws IOException { this.tag.readFields(in); //加入此部分代码,否则,可能报空指针异常 String temp=in.readUTF(); if (this.data == null|| !this.data.getClass().getName().equals(temp)) { try { this.data = (Writable) ReflectionUtils.newInstance( Class.forName(temp), null); } catch (ClassNotFoundException e) { e.printStackTrace(); } } this.data.readFields(in); } } public int run(String[] args) throws Exception { Configuration conf = getConf(); JobConf job = new JobConf(conf, DataJoin.class); job.set("mapred.job.tracker","192.168.75.130:9001"); ////读取person中的数据字段 job.setJar("tt.jar"); job.setJarByClass(DataJoin.class); System.out.println("模式: "+job.get("mapred.job.tracker"));; String path="hdfs://192.168.75.130:9000/root/outputjoindb"; FileSystem fs=FileSystem.get(conf); Path p=new Path(path); if(fs.exists(p)){ fs.delete(p, true); System.out.println("输出路径存在,已删除!"); } Path in = new Path("hdfs://192.168.75.130:9000/root/inputjoindb"); // Path out = new Path(args[1]); FileInputFormat.setInputPaths(job, in); FileOutputFormat.setOutputPath(job, p); job.setJobName("cee"); job.setMapperClass(MapClass.class); job.setReducerClass(Reduce.class); job.setInputFormat(TextInputFormat.class); job.setOutputFormat(TextOutputFormat.class); job.setOutputKeyClass(Text.class); job.setOutputValueClass(TaggedWritable.class); job.set("mapred.textoutputformat.separator", ","); JobClient.runJob(job); return 0; } public static void main(String[] args) throws Exception { int res = ToolRunner.run(new Configuration(), new DataJoin(), args); System.exit(res); } }
运行,日志
模式: 192.168.75.130:9001 输出路径存在,已删除! INFO - NativeCodeLoader.<clinit>(43) | Loaded the native-hadoop library WARN - LoadSnappy.<clinit>(46) | Snappy native library not loaded INFO - FileInputFormat.listStatus(199) | Total input paths to process : 2 INFO - JobClient.monitorAndPrintJob(1380) | Running job: job_201404222310_0025 INFO - JobClient.monitorAndPrintJob(1393) | map 0% reduce 0% INFO - JobClient.monitorAndPrintJob(1393) | map 33% reduce 0% INFO - JobClient.monitorAndPrintJob(1393) | map 100% reduce 0% INFO - JobClient.monitorAndPrintJob(1393) | map 100% reduce 33% INFO - JobClient.monitorAndPrintJob(1393) | map 100% reduce 100% INFO - JobClient.monitorAndPrintJob(1448) | Job complete: job_201404222310_0025 INFO - Counters.log(585) | Counters: 30 INFO - Counters.log(587) | Job Counters INFO - Counters.log(589) | Launched reduce tasks=1 INFO - Counters.log(589) | SLOTS_MILLIS_MAPS=14335 INFO - Counters.log(589) | Total time spent by all reduces waiting after reserving slots (ms)=0 INFO - Counters.log(589) | Total time spent by all maps waiting after reserving slots (ms)=0 INFO - Counters.log(589) | Launched map tasks=3 INFO - Counters.log(589) | Data-local map tasks=3 INFO - Counters.log(589) | SLOTS_MILLIS_REDUCES=9868 INFO - Counters.log(587) | File Input Format Counters INFO - Counters.log(589) | Bytes Read=207 INFO - Counters.log(587) | File Output Format Counters INFO - Counters.log(589) | Bytes Written=172 INFO - Counters.log(587) | FileSystemCounters INFO - Counters.log(589) | FILE_BYTES_READ=837 INFO - Counters.log(589) | HDFS_BYTES_READ=513 INFO - Counters.log(589) | FILE_BYTES_WRITTEN=221032 INFO - Counters.log(589) | HDFS_BYTES_WRITTEN=172 INFO - Counters.log(587) | Map-Reduce Framework INFO - Counters.log(589) | Map output materialized bytes=849 INFO - Counters.log(589) | Map input records=8 INFO - Counters.log(589) | Reduce shuffle bytes=849 INFO - Counters.log(589) | Spilled Records=16 INFO - Counters.log(589) | Map output bytes=815 INFO - Counters.log(589) | Total committed heap usage (bytes)=496644096 INFO - Counters.log(589) | CPU time spent (ms)=2080 INFO - Counters.log(589) | Map input bytes=187 INFO - Counters.log(589) | SPLIT_RAW_BYTES=306 INFO - Counters.log(589) | Combine input records=0 INFO - Counters.log(589) | Reduce input records=8 INFO - Counters.log(589) | Reduce input groups=4 INFO - Counters.log(589) | Combine output records=0 INFO - Counters.log(589) | Physical memory (bytes) snapshot=623570944 INFO - Counters.log(589) | Reduce output records=4 INFO - Counters.log(589) | Virtual memory (bytes) snapshot=2908262400 INFO - Counters.log(589) | Map output records=8
运行结果,如下图所示:
可以看出,MR正确的完成了join操作,需要注意的是Reduce侧连接的不足之处:
之所以会存在reduce join这种方式,我们可以很明显的看出原:因为整体数据被分割了,每个map task只处理一部分数据而不能够获取到所有需要的join字段,因此我们需要在讲join key作为reduce端的分组将所有join key相同的记录集中起来进行处理,所以reduce join这种方式就出现了。这种方式的缺点很明显就是会造成map和reduce端也就是shuffle阶段出现大量的数据传输,效率很低。
另外一点需要注意的是,散仙在eclipse里进行调试,Local模式下会报异常,建议提交到hadoop的测试集群上进行测试。
发表评论
-
Apache Flink在阿里的使用(译)
2019-02-21 21:18 1059Flink是未来大数据实时 ... -
计算机图形处理的一些知识
2018-04-25 17:46 1191最近在搞opencv来做一些 ... -
如何在kylin中构建一个cube
2017-07-11 19:06 1208前面的文章介绍了Apache Kylin的安装及数据仓 ... -
Apache Kylin的入门安装
2017-06-27 21:27 2112Apache Kylin™是一个开源的分布式分析引擎,提供 ... -
ES-Hadoop插件介绍
2017-04-27 18:07 1947上篇文章,写了使用spark集成es框架,并向es写入数据,虽 ... -
如何在Scala中读取Hadoop集群上的gz压缩文件
2017-04-05 18:51 2090存在Hadoop集群上的文件,大部分都会经过压缩,如果是压缩 ... -
如何收集项目日志统一发送到kafka中?
2017-02-07 19:07 2749上一篇(http://qindongliang.iteye. ... -
Hue+Hive临时目录权限不够解决方案
2016-06-14 10:40 4641安装Hue后,可能会分配多个账户给一些业务部门操作hive,虽 ... -
Hadoop的8088页面失效问题
2016-03-31 11:21 4381前两天重启了测试的hadoop集群,今天访问集群的8088任 ... -
Hadoop+Hbase集群数据迁移问题
2016-03-23 21:00 2478数据迁移或备份是任何 ... -
如何监控你的Hadoop+Hbase集群?
2016-03-21 16:10 4834前言 监控hadoop的框架 ... -
Logstash与Kafka集成
2016-02-24 18:44 11508在ELKK的架构中,各个框架的角色分工如下: Elastic ... -
Kakfa集群搭建
2016-02-23 15:36 2592先来整体熟悉下Kafka的一些概念和架构 (一)什么是Ka ... -
大数据日志收集框架之Flume入门
2016-02-02 14:25 4147Flume是Cloudrea公司开源的一款优秀的日志收集框架 ... -
Apache Tez0.7编译笔记
2016-01-15 16:33 2430目前最新的Tez版本是0.8,但还不是稳定版,所以大家还 ... -
Bug死磕之hue集成的oozie+pig出现资源任务死锁问题
2016-01-14 15:52 3758这两天,打算给现有的 ... -
Hadoop2.7.1和Hbase0.98添加LZO压缩
2016-01-04 17:46 25521,执行命令安装一些依赖组件 yum install -y ... -
Hadoop2.7.1配置NameNode+ResourceManager高可用原理分析
2015-11-11 19:51 3129关于NameNode高可靠需要配置的文件有core-site ... -
设置Hadoop+Hbase集群pid文件存储位置
2015-10-20 13:40 2782有时候,我们对运行几 ... -
Hadoop+Maven项目打包异常
2015-08-11 19:36 1505先简单说下业务:有一个单独的模块,可以在远程下载Hadoop上 ...
相关推荐
19、Join操作map side join 和 reduce side join 网址:https://blog.csdn.net/chenwewi520feng/article/details/130455477 本文介绍mapreduce的join操作。 本文前提是hadoop可以正常使用。 本文分为3个部分介绍,即...
展示使用MR方式实现表连接的代码示例。利用HIVE PIG之类的高层工具也可以实现,本代码旨在展示手工连接的流程
源码 博文链接:https://shihlei.iteye.com/blog/2263757
The Joins query by using Hadoop and map reduce
Hadoop Map Reduce教程,介绍hadoop map/reduce框架的各个方面
hadoop-0.21.0-datajoin.jar
hadoop的map reduce 学习手册,很实用
hadoop map reduce 的中文简易教程,能轻松帮助普通用户不需了解太多hadoop底层知识就能实现分布式编程,很好的入门教程。
人脸识别,车辆识别,一人一档,一车一档 hadoop map reduce hbase
Joins in Hadoop has always been a problem for its users: the Map/Reduce framework seems to be specifically designed for group-by aggregation tasks rather than across-table op- erations; on the other ...
使用hadoop实现WordCount详细实验报告,配有环境变量配置截图以及实验运行及结果详细过程描述与截图
Hadoop Map Reduce 教程.doc
hadoop开发文档
hadoop中map/reduce自学资料合集
Hadoop Map-Reduce教程,hadoop,mapreduce
使用Hadoop Map Reduce分析股票市场 如何运行程序? 首先在您的系统中安装Hadoop。 请按照以下步骤进行安装 然后开始执行给定的命令 cd hadoop-3.2.2 / sbin ./start-dfs.sh ./start-yarn.sh jps 导出HADOOP_...
讲述了Windows平台的Hadoop安装... 最后,以最简单的求和为例,剖析Hadoop的Map/Reduce工作机制,对于初学Hadoop及Map/Reduce的读者有很大的帮助。相信通过最简单的求和为例,读者可步入Hadoop的Map/Reduce开发者行列。
Hadoop学习总结之三:Map-Reduce入门
Hadoop学习总结之四:Map-Reduce的过程解析
Hadoop使用常见问题以及解决方法,简单实用