- 浏览: 2144973 次
- 性别:
- 来自: 北京
文章分类
- 全部博客 (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开发之第八步之渡劫初期(八)
使用Hadoop里面的MapReduce来处理海量数据是非常简单方便的,但有时候我们的应用程序,往往需要多个MR作业,来计算结果,比如说一个最简单的使用MR提取海量搜索日志的TopN的问题,注意,这里面,其实涉及了两个MR作业,第一个是词频统计,第两个是排序求TopN,这显然是需要两个MapReduce作业来完成的。其他的还有,比如一些数据挖掘类的作业,常常需要迭代组合好几个作业才能完成,这类作业类似于DAG类的任务,各个作业之间是具有先后,或相互依赖的关系,比如说,这一个作业的输入,依赖上一个作业的输出等等。
在Hadoop里实际上提供了,JobControl类,来组合一个具有依赖关系的作业,在新版的API里,又新增了ControlledJob类,细化了任务的分配,通过这两个类,我们就可以轻松的完成类似DAG作业的模式,这样我们就可以通过一个提交来完成原来需要提交2次的任务,大大简化了任务的繁琐度。具有依赖式的作业提交后,hadoop会根据依赖的关系,先后执行的job任务,每个任务的运行都是独立的。
下面来看下散仙的例子,组合一个词频统计+排序的作业,测试数据如下:
代码如下:
运行日志如下:
处理的结果如下:
可以看出,结果是正确的。程序运行成功,上面只是散仙测的2个MapReduce作业的组合,更多的组合其实和上面的一样。
总结:在配置多个作业时,Job的配置尽量分离单独写,不要轻易拷贝修改,这样很容易出错的,散仙在配置的时候,就是拷贝了一个,结果因为少修改了一个地方,在运行时候一直报错,最后才发现,原来少改了某个地方,这一点需要注意一下。
在Hadoop里实际上提供了,JobControl类,来组合一个具有依赖关系的作业,在新版的API里,又新增了ControlledJob类,细化了任务的分配,通过这两个类,我们就可以轻松的完成类似DAG作业的模式,这样我们就可以通过一个提交来完成原来需要提交2次的任务,大大简化了任务的繁琐度。具有依赖式的作业提交后,hadoop会根据依赖的关系,先后执行的job任务,每个任务的运行都是独立的。
下面来看下散仙的例子,组合一个词频统计+排序的作业,测试数据如下:
秦东亮;72 秦东亮;34 秦东亮;100 三劫;899 三劫;32 三劫;1 a;45 b;567 b;12
代码如下:
package com.qin.test.hadoop.jobctrol; import java.io.IOException; import org.apache.hadoop.fs.FileSystem; import org.apache.hadoop.fs.Path; import org.apache.hadoop.io.IntWritable; import org.apache.hadoop.io.LongWritable; import org.apache.hadoop.io.Text; import org.apache.hadoop.io.WritableComparator; import org.apache.hadoop.mapred.JobConf; import org.apache.hadoop.mapreduce.Job; import org.apache.hadoop.mapreduce.Mapper; import org.apache.hadoop.mapreduce.Reducer; import org.apache.hadoop.mapreduce.lib.input.FileInputFormat; import org.apache.hadoop.mapreduce.lib.jobcontrol.ControlledJob; import org.apache.hadoop.mapreduce.lib.jobcontrol.JobControl; import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat; /** * Hadoop的版本是1.2的 * JDK环境1.6 * 使用ControlledJob+JobControl新版API * 完成组合式任务 * 第一个任务是统计词频 * 第二个任务是降序排序 * * 如果使用MapReduce作业来完成的话,则需要跑2个MR任务 * 但是如果我们使用了JobControl+ControlledJob就可以在 * 一个类里面完成类型的DAG依赖式的作业 * * * @author qindongliang * * * * ***/ public class MyHadoopControl { /*** * *MapReduce作业1的Mapper * *LongWritable 1 代表输入的key值,默认是文本的位置偏移量 *Text 2 每行的具体内容 *Text 3 输出的Key类型 *Text 4 输出的Value类型 * * */ private static class SumMapper extends Mapper<LongWritable, Text, Text, IntWritable>{ private Text t=new Text(); private IntWritable one=new IntWritable(1); /** * * map阶段输出词频 * * * **/ @Override protected void map(LongWritable key, Text value,Context context) throws IOException, InterruptedException { String data=value.toString(); String words[]=data.split(";"); if(words[0].trim()!=null){ t.set(" "+words[0]);//赋值K one.set(Integer.parseInt(words[1])); context.write(t, one); } } } /** * MapReduce作业1的Reducer * 负责词频累加,并输出 * * **/ private static class SumReduce extends Reducer<Text, IntWritable, IntWritable, Text>{ //存储词频对象 private IntWritable iw=new IntWritable(); @Override protected void reduce(Text key, Iterable<IntWritable> value,Context context) throws IOException, InterruptedException { int sum=0; for(IntWritable count:value){ sum+=count.get();//累加词频 } iw.set(sum);//设置词频 context.write(iw, key);//输出数据 } } /** * MapReduce作业2排序的Mapper * * **/ private static class SortMapper extends Mapper<LongWritable, Text, IntWritable, Text>{ IntWritable iw=new IntWritable();//存储词频 private Text t=new Text();//存储文本 @Override protected void map(LongWritable key, Text value,Context context)throws IOException, InterruptedException { String words[]=value.toString().split(" "); System.out.println("数组的长度: "+words.length); System.out.println("Map读入的文本: "+value.toString()); System.out.println("=====> "+words[0]+" =====>"+words[1]); if(words[0]!=null){ iw.set(Integer.parseInt(words[0].trim())); t.set(words[1].trim()); context.write(iw, t);//map阶段输出,默认按key排序 } } } /** * MapReduce作业2排序的Reducer * * **/ private static class SortReduce extends Reducer<IntWritable, Text, Text, IntWritable>{ /** * * 输出排序内容 * * **/ @Override protected void reduce(IntWritable key, Iterable<Text> value,Context context) throws IOException, InterruptedException { for(Text t:value){ context.write(t, key);//输出排好序后的K,V } } } /*** * 排序组件,在排序作业中,需要使用 * 按key的降序排序 * * **/ public static class DescSort extends WritableComparator{ public DescSort() { super(IntWritable.class,true);//注册排序组件 } @Override public int compare(byte[] arg0, int arg1, int arg2, byte[] arg3, int arg4, int arg5) { return -super.compare(arg0, arg1, arg2, arg3, arg4, arg5);//注意使用负号来完成降序 } @Override public int compare(Object a, Object b) { return -super.compare(a, b);//注意使用负号来完成降序 } } /** * 驱动类 * * **/ public static void main(String[] args)throws Exception { JobConf conf=new JobConf(MyHadoopControl.class); conf.set("mapred.job.tracker","192.168.75.130:9001"); conf.setJar("tt.jar"); System.out.println("模式: "+conf.get("mapred.job.tracker"));; /** * *作业1的配置 *统计词频 * * **/ Job job1=new Job(conf,"Join1"); job1.setJarByClass(MyHadoopControl.class); job1.setMapperClass(SumMapper.class); job1.setReducerClass(SumReduce.class); job1.setMapOutputKeyClass(Text.class);//map阶段的输出的key job1.setMapOutputValueClass(IntWritable.class);//map阶段的输出的value job1.setOutputKeyClass(IntWritable.class);//reduce阶段的输出的key job1.setOutputValueClass(Text.class);//reduce阶段的输出的value //加入控制容器 ControlledJob ctrljob1=new ControlledJob(conf); ctrljob1.setJob(job1); FileInputFormat.addInputPath(job1, new Path("hdfs://192.168.75.130:9000/root/input")); FileSystem fs=FileSystem.get(conf); Path op=new Path("hdfs://192.168.75.130:9000/root/op"); if(fs.exists(op)){ fs.delete(op, true); System.out.println("存在此输出路径,已删除!!!"); } FileOutputFormat.setOutputPath(job1, op); /**========================================================================*/ /** * *作业2的配置 *排序 * * **/ Job job2=new Job(conf,"Join2"); job2.setJarByClass(MyHadoopControl.class); //job2.setInputFormatClass(TextInputFormat.class); job2.setMapperClass(SortMapper.class); job2.setReducerClass(SortReduce.class); job2.setSortComparatorClass(DescSort.class);//按key降序排序 job2.setMapOutputKeyClass(IntWritable.class);//map阶段的输出的key job2.setMapOutputValueClass(Text.class);//map阶段的输出的value job2.setOutputKeyClass(Text.class);//reduce阶段的输出的key job2.setOutputValueClass(IntWritable.class);//reduce阶段的输出的value //作业2加入控制容器 ControlledJob ctrljob2=new ControlledJob(conf); ctrljob2.setJob(job2); /*** * * 设置多个作业直接的依赖关系 * 如下所写: * 意思为job2的启动,依赖于job1作业的完成 * * **/ ctrljob2.addDependingJob(ctrljob1); //输入路径是上一个作业的输出路径 FileInputFormat.addInputPath(job2, new Path("hdfs://192.168.75.130:9000/root/op/part*")); FileSystem fs2=FileSystem.get(conf); Path op2=new Path("hdfs://192.168.75.130:9000/root/op2"); if(fs2.exists(op2)){ fs2.delete(op2, true); System.out.println("存在此输出路径,已删除!!!"); } FileOutputFormat.setOutputPath(job2, op2); // System.exit(job2.waitForCompletion(true) ? 0 : 1); /**====================================================================***/ /** * * 主的控制容器,控制上面的总的两个子作业 * * **/ JobControl jobCtrl=new JobControl("myctrl"); //ctrljob1.addDependingJob(ctrljob2);// job2在job1完成后,才可以启动 //添加到总的JobControl里,进行控制 jobCtrl.addJob(ctrljob1); jobCtrl.addJob(ctrljob2); //在线程启动 Thread t=new Thread(jobCtrl); t.start(); while(true){ if(jobCtrl.allFinished()){//如果作业成功完成,就打印成功作业的信息 System.out.println(jobCtrl.getSuccessfulJobList()); jobCtrl.stop(); break; } if(jobCtrl.getFailedJobList().size()>0){//如果作业失败,就打印失败作业的信息 System.out.println(jobCtrl.getFailedJobList()); jobCtrl.stop(); break; } } } }
运行日志如下:
模式: 192.168.75.130:9001 存在此输出路径,已删除!!! 存在此输出路径,已删除!!! WARN - JobClient.copyAndConfigureFiles(746) | Use GenericOptionsParser for parsing the arguments. Applications should implement Tool for the same. INFO - FileInputFormat.listStatus(237) | Total input paths to process : 1 WARN - NativeCodeLoader.<clinit>(52) | Unable to load native-hadoop library for your platform... using builtin-java classes where applicable WARN - LoadSnappy.<clinit>(46) | Snappy native library not loaded WARN - JobClient.copyAndConfigureFiles(746) | Use GenericOptionsParser for parsing the arguments. Applications should implement Tool for the same. INFO - FileInputFormat.listStatus(237) | Total input paths to process : 1 [job name: Join1 job id: myctrl0 job state: SUCCESS job mapred id: job_201405092039_0001 job message: just initialized job has no depending job: , job name: Join2 job id: myctrl1 job state: SUCCESS job mapred id: job_201405092039_0002 job message: just initialized job has 1 dependeng jobs: depending job 0: Join1 ]
处理的结果如下:
三劫 932 b 579 秦东亮 206 a 45
可以看出,结果是正确的。程序运行成功,上面只是散仙测的2个MapReduce作业的组合,更多的组合其实和上面的一样。
总结:在配置多个作业时,Job的配置尽量分离单独写,不要轻易拷贝修改,这样很容易出错的,散仙在配置的时候,就是拷贝了一个,结果因为少修改了一个地方,在运行时候一直报错,最后才发现,原来少改了某个地方,这一点需要注意一下。
发表评论
-
Apache Flink在阿里的使用(译)
2019-02-21 21:18 1053Flink是未来大数据实时 ... -
计算机图形处理的一些知识
2018-04-25 17:46 1187最近在搞opencv来做一些 ... -
如何在kylin中构建一个cube
2017-07-11 19:06 1204前面的文章介绍了Apache Kylin的安装及数据仓 ... -
Apache Kylin的入门安装
2017-06-27 21:27 2105Apache Kylin™是一个开源的分布式分析引擎,提供 ... -
ES-Hadoop插件介绍
2017-04-27 18:07 1942上篇文章,写了使用spark集成es框架,并向es写入数据,虽 ... -
如何在Scala中读取Hadoop集群上的gz压缩文件
2017-04-05 18:51 2083存在Hadoop集群上的文件,大部分都会经过压缩,如果是压缩 ... -
如何收集项目日志统一发送到kafka中?
2017-02-07 19:07 2742上一篇(http://qindongliang.iteye. ... -
Hue+Hive临时目录权限不够解决方案
2016-06-14 10:40 4631安装Hue后,可能会分配多个账户给一些业务部门操作hive,虽 ... -
Hadoop的8088页面失效问题
2016-03-31 11:21 4367前两天重启了测试的hadoop集群,今天访问集群的8088任 ... -
Hadoop+Hbase集群数据迁移问题
2016-03-23 21:00 2471数据迁移或备份是任何 ... -
如何监控你的Hadoop+Hbase集群?
2016-03-21 16:10 4828前言 监控hadoop的框架 ... -
Logstash与Kafka集成
2016-02-24 18:44 11500在ELKK的架构中,各个框架的角色分工如下: Elastic ... -
Kakfa集群搭建
2016-02-23 15:36 2587先来整体熟悉下Kafka的一些概念和架构 (一)什么是Ka ... -
大数据日志收集框架之Flume入门
2016-02-02 14:25 4140Flume是Cloudrea公司开源的一款优秀的日志收集框架 ... -
Apache Tez0.7编译笔记
2016-01-15 16:33 2425目前最新的Tez版本是0.8,但还不是稳定版,所以大家还 ... -
Bug死磕之hue集成的oozie+pig出现资源任务死锁问题
2016-01-14 15:52 3750这两天,打算给现有的 ... -
Hadoop2.7.1和Hbase0.98添加LZO压缩
2016-01-04 17:46 25491,执行命令安装一些依赖组件 yum install -y ... -
Hadoop2.7.1配置NameNode+ResourceManager高可用原理分析
2015-11-11 19:51 3122关于NameNode高可靠需要配置的文件有core-site ... -
设置Hadoop+Hbase集群pid文件存储位置
2015-10-20 13:40 2778有时候,我们对运行几 ... -
Hadoop+Maven项目打包异常
2015-08-11 19:36 1499先简单说下业务:有一个单独的模块,可以在远程下载Hadoop上 ...
相关推荐
可以帮助开发者自动安装Android SDK、Android NDK、Java开发工具包(JDK)以及Flutter SDK,这些都是移动应用开发(特别是Android和Flutter应用)所必需的。 脚本首先定义了一些变量,用于存储工具的安装路径和版本。然后,定义了一个install_tool函数,用于检查工具是否已安装,如果没有安装则执行安装命令。 接着,脚本定义了install_android_sdk、install_android_ndk、install_java_jdk、set_java_env和install_flutter_sdk等函数,分别用于安装Android SDK、Android NDK、Java JDK、设置Java环境变量以及安装Flutter SDK。 在主程序部分,脚本按照顺序调用这些函数来安装和配置各个工具。安装完成后,脚本会输出一条消息表示配置已完成。
AI Agent智能应用从0到1定制开发 关键特点: 自动化:AI Agent可以自动执行重复性任务,提高效率并减少人力成本。 智能决策:基于复杂的算法,AI Agent能够进行决策支持,分析数据并提供洞察。 自然语言处理:许多AI Agent具备理解和生成自然语言的能力,使其能够与人类用户进行交流。 机器学习:AI Agent可以利用机器学习技术从经验中学习,不断优化其性能。 个性化:AI Agent能够根据用户的行为和偏好提供个性化的体验。 可扩展性:AI Agent可以设计成模块化,方便扩展新功能或适应不同规模的需求。 实时响应:AI Agent能够提供快速的实时响应,满足紧急任务的需求。
魔方PPT模板04_动态清新绿色学术答辩模板.pptx
Tomcat是一个广泛使用的开源Web服务器和Servlet容器,它是由Apache软件基金会(ASF)主持的一个项目。Tomcat实现了Java Servlet、JavaServer Pages(JSP)、Java Expression Language(JSTL)和WebSocket等技术规范,并提供了一个用于运行Java Web应用程序的运行时环境。由于其稳定性、可扩展性和灵活性,Tomcat已成为许多企业级Java Web应用程序的首选服务器。 Tomcat最初是由Sun Microsystems在1999年作为Servlet API 2.2和JSP 1.1规范的参考实现而开发的。自那时以来,Tomcat不断发展壮大,成为了一个功能强大且易于使用的Web服务器和Servlet容器。Tomcat的开源性质和广泛的支持使得它成为了许多开发者和企业的首选。
用法链接:https://menghui666.blog.csdn.net/article/details/137888208?spm=1001.2014.3001.5502 基于Qt+C++实现的各种炫酷的样式表,如单选、多选、按钮、日历、表格、下拉框、滚轮等,+源码 基于Qt+C++实现的各种炫酷的样式表,如单选、多选、按钮、日历、表格、下拉框、滚轮等,+源码 基于Qt+C++实现的各种炫酷的样式表,如单选、多选、按钮、日历、表格、下拉框、滚轮等,+源码
罗茨泵设计sw17可编辑含工程图设计图纸.7z
计算机思维训练+C语言
基于JAVA毕业设计-JAVA基于遗传算法的中药药对挖掘系统的设计与实现(源代码+论文).rar 用数据挖掘技术研究了中药方剂配伍的规律。主要工作:分析了关联规则存在的问题,引入双向关联规则的概念;介绍了遗传算法的基本原理,研究了遗传算法在数据挖掘中的应用;将方剂库转换为位图矩阵,大大提高搜索效率;开发了一个基于遗传算法的中药药对药组挖掘系统。论文组织如下:介绍了研究背景和意义;阐述了相关的理论基础;提出了系统的设计方案;详细展示了基于遗传算法的双向关联规则挖掘系统的实现过程,包括位图矩阵的实现,个体的编码方法,适应度函数的设计,规则的提取,选择、交叉、变异等遗传操作的实现等;利用脾胃类方剂库对系统进行了测试,并对测试结果进行了分析。结果证明:该系统能够快速高效地从方剂库中找出具有重要意义的药对药组,对中医药的研究发展有一定意义。
C常用算法程序集数值计算数值积分提取方式是百度网盘分享地址
raun内外盒组装包装设备含工程图sw17可编辑设计图纸.7z
四路仿PLC继电器控制板原理图+教程+源代码+下位机源代码+上位机源代码+其它技术资料.7z
数据来源:中经数据库 数据范围:各个省份的区县财政收入即一般公共预算收入、税收收入 (一般财政收入即一般公共预算收入的完整度较高。税收收入一般50%的区县会有数据) 数据年度区间:2000-2023(具体看文件名上的年度区间) 珍贵数据,包含了各省所有的区、县、县级市哦,很难找到的哦
2021年全国职业院校技能大赛模块ALinux评分标准
tomcatCentos7下tomcat的安装.zip
绚彩小清新毕业答辩模板.ppt
毕业设计-微信小程序选座系统源码.zip文件 毕业设计-微信小程序选座系统源码.zip文件毕业设计-微信小程序选座系统源码.zip文件毕业设计-微信小程序选座系统源码.zip文件毕业设计-微信小程序选座系统源码.zip文件毕业设计-微信小程序选座系统源码.zip文件毕业设计-微信小程序选座系统源码.zip文件毕业设计-微信小程序选座系统源码.zip文件毕业设计-微信小程序选座系统源码.zip文件毕业设计-微信小程序选座系统源码.zip文件毕业设计-微信小程序选座系统源码.zip文件毕业设计-微信小程序选座系统源码.zip文件毕业设计-微信小程序选座系统源码.zip文件毕业设计-微信小程序选座系统源码.zip文件毕业设计-微信小程序选座系统源码.zip文件毕业设计-微信小程序选座系统源码.zip文件毕业设计-微信小程序选座系统源码.zip文件毕业设计-微信小程序选座系统源码.zip文件毕业设计-微信小程序选座系统源码.zip文件毕业设计-微信小程序选座系统源码.zip文件毕业设计-微信小程序选座系统源码.zip文件毕业设计-微信小程序选座系统源码.zip文件毕业设计-微信小程序选座系
Android Studio移动应用开发PPT详解《微课版》
DataV Utils 文件
计算机毕业设计,含源码
机器学习是一种人工智能(AI)的子领域,致力于研究如何利用数据和算法让计算机系统具备学习能力,从而能够自动地完成特定任务或者改进自身性能。机器学习的核心思想是让计算机系统通过学习数据中的模式和规律来实现目标,而不需要显式地编程。 机器学习应用非常广泛,包括但不限于以下领域: 图像识别和计算机视觉: 机器学习在图像识别、目标检测、人脸识别、图像分割等方面有着广泛的应用。例如,通过深度学习技术,可以训练神经网络来识别图像中的对象、人脸或者场景,用于智能监控、自动驾驶、医学影像分析等领域。 自然语言处理: 机器学习在自然语言处理领域有着重要的应用,包括文本分类、情感分析、机器翻译、语音识别等。例如,通过深度学习模型,可以训练神经网络来理解和生成自然语言,用于智能客服、智能助手、机器翻译等场景。 推荐系统: 推荐系统利用机器学习算法分析用户的行为和偏好,为用户推荐个性化的产品或服务。例如,电商网站可以利用机器学习算法分析用户的购买历史和浏览行为,向用户推荐感兴趣的商品。 预测和预测分析: 机器学习可以用于预测未来事件的发生概率或者趋势。例如,金融领域可以利用机器学习算法进行股票价格预测、信用评分、欺诈检测等。 医疗诊断和生物信息学: 机器学习在医疗诊断、药物研发、基因组学等领域有着重要的应用。例如,可以利用机器学习算法分析医学影像数据进行疾病诊断,或者利用机器学习算法分析基因数据进行疾病风险预测。 智能交通和物联网: 机器学习可以应用于智能交通系统、智能城市管理和物联网等领域。例如,可以利用机器学习算法分析交通数据优化交通流量,或者利用机器学习算法分析传感器数据监测设备状态。 以上仅是机器学习应用的一部分,随着机器学习技术的不断发展和应用场景的不断拓展,机器学习在各个领域都有着重要的应用价值,并且正在改变我们的生活和工作方式。