问题描述
我们的数据分析平台是单一的Map/Reduce过程,由于半年来不断地增加需求,导致了问题已经不是那么地简单,特别是在Reduce阶段,一些大对象会常驻内存。因此越来越顶不住压力了,当前内存问题已经是最大的问题,每个Map占用5G,每个Reduce占用9G!直接导致当数据分析平台运行时,集群处于资源匮乏状态。
因此,在不改变业务数据计算的条件下,将单一的Map/Reduce过程分解成2个阶段。这个时候,需求就相对来说比较复杂,将第一阶段的Reduce结果输出至HDFS,作为第二阶段的输入。
其基本过程图很简单如下所示:
我们可以使用启动两个Job,在第一个阶段Job完成之后,再进行第二阶段Job的执行。但是更好的方式是使用Hadoop中提供的JobControl工具,这个工具可以加入多个等待执行的子Job,并定义其依赖关系,决定执行的先后顺序。
JobControl jobControl = new JobControl(GROUP_NAME); JobConf phase1JobConf = Phase1Main.getJobConf(getConf(), jsonConfigFilePhase1, reduceCountOne); Job phase1Job = new Job(phase1JobConf); jobControl.addJob(phase1Job); JobConf phase2JobConf = Phase2Main.getJobConf(getConf(), jsonConfigFilePhase2, reduceCountTwo); Job phase2Job = new Job(phase2JobConf); jobControl.addJob(phase2Job); phase2Job.addDependingJob(phase1Job); jobControl.run(); return jobControl.getFailedJobs() == null || jobControl.getFailedJobs().isEmpty() ? 0 : 1;
正如代码所示,可以使用job的addDependingJob(JobConf)方法来定义其依赖关系。
但是这种方式有一个非常大的缺点,如果中间数据结果过大,将其放置在HDFS上是非常浪费磁盘资源,同时也带来后续过多的I/O操作,包括第一阶段的写磁盘和第二阶段的读磁盘(而且本身中间结果数据也没有什么太大用途)。
经过查阅,在Hadoop中,一个Job可以按顺序执行多个mapper对数据进行前期的处理,再进行Reduce,Reduce执行完成后,还可以继续执行多个Mapper,形成一个处理链结构,这样的Job是不会存储中间结果的,大大减少了磁盘I/O操作。
但这种方式也对map/reduce程序有个要求,就是只能存在一个Partition规则,因为整个链条中只会存在一次Reduce操作。前文介绍的那两个阶段的Partition规则如果不一致,是不能改造成这种方式的。
这种方式的大致流程图如下:
由于我们的分析程序中,第二步就需要根据一定的规则进行聚集,因此第二步就需要进行Reduce,将原来第四步的Reduce阶段强行改造成Map阶段。注意,Map阶段之间互相传递数据时,其数量是固定的,而且不会进行聚集(Reduce)操作,还是需要按照流的方式进行处理,因此最好要先排序。单个Map的结果只会传递给特定的单个下个步骤的Map端。
在ChainMain类中会执行这种方式,需要借助于ChainMapper和ChainReducer两个Hadoop中提供的类:
String finalJobName = TongCommonConstants.JOB_NAME + jobNameSuffix; jobConf.setJobName(finalJobName); jobConf.setInputFormat(RawLogInputFormat.class); jobConf.setPartitionerClass(Phase1Partitioner.class); jobConf.setNumReduceTasks(reduceCountTwo); jobConf.set(TongCommonConstants.DIC_INFO, jsonConfigFile); DicInfoManager.getInstance().readDicManager(jobConf, jsonConfigFile); String yesterdayOutDir = DicInfoManager.getInstance().getDicManager().getPrevious_day_output_path(); JobConf phase1JobConf = getJobConf(jsonConfigFile, yesterdayOutDir); ChainMapper.addMapper(jobConf, Phase1Mapper.class, Text.class, History.class, Phase1KeyDecorator.class, BytesWritable.class, true, phase1JobConf); JobConf phase2ReducerConf = getJobConf(jsonConfigFile, yesterdayOutDir); ChainReducer.setReducer(jobConf, Phase1Reducer.class, Phase1KeyDecorator.class, BytesWritable.class, Text.class, Text.class, true, phase2ReducerConf); JobConf phase3ChainJobConf = getJobConf(jsonConfigFile, yesterdayOutDir); ChainReducer.addMapper(jobConf, Phase3ChainMapper.class, Text.class, Text.class, Phase2KeyDecorator.class, BytesWritable.class, true, phase3ChainJobConf); JobConf phase4ChainJobConf = getJobConf(jsonConfigFile, yesterdayOutDir); ChainReducer.addMapper(jobConf, Phase4ChainMapper.class, Phase2KeyDecorator.class, BytesWritable.class, Text.class, Text.class, true, phase4ChainJobConf); RunningJob runningJob = JobClient.runJob(jobConf); runningJob.waitForCompletion(); return runningJob.isSuccessful() ? 0 : 1;
经过这种方式的改造后,对原有程序的影响最小,因为不需要定义中间结果存储地址,当然也不需要定义第二阶段的配置文件。
新手比较容易犯的一个错误是,Reducer后面的map步骤要使用ChainReducer.addMapper方法而不是ChainMapper.addMapper方法,否则会抱下面的异常,我就在这个上面栽了跟头,查了很久。
Exception in thread "main" java.lang.IllegalArgumentException: The specified Mapper input key class does not match the previous Mapper's output key class. at org.apache.hadoop.mapreduce.lib.chain.Chain.validateKeyValueTypes(Chain.java:695) at org.apache.hadoop.mapred.lib.Chain.addMapper(Chain.java:104)
Hadoop工作流中的JobControl
很多情况下,用户编写的作业比较复杂,相互之间存在依赖关系,这种可以用有向图表示的依赖关系称之为“工作流”。
JobControl是由两个类组成:Job和JobControl,Job的状态转移图如下:
作业在刚开始的时候处于Waiting状态,如果没有依赖作业或者所有依赖作业都已经完成的情况下,进入Ready状态;一旦进入Ready状态,则作业可被提交到Hadoop集群上运行,并进入Running状态,根据作业的运行情况,可能进入Success或Failed状态。需要注意的是,如果一个作业的依赖作业失败,则该作业也会失败,后续的所有作业也都会失败。
JobControl封装了一系列MapReduce作业及其对应的依赖关系,它将处于不同状态的作业放入不同的哈希表,按照Job的状态转移图转移作业,直到所有作业运行完成。在实现的时候,JobControl包含一个线程用于周期性地监控和更新各个作业的运行状态,调度依赖作业运行完成的作业,提交Ready状态的作业等。
相关推荐
7. **工具集成**:有许多开源工具可以帮助我们远程提交和管理Hadoop作业,如Hadoop命令行工具、Hadoop的Web UI、Apache Oozie工作流管理系统等。这些工具提供了方便的接口,使开发者能便捷地与集群交互。 8. **安全...
标题中的“在solr文献检索中用map/reduce”指的是使用Apache Solr,一个流行的开源搜索引擎,结合Hadoop的MapReduce框架来处理大规模的分布式搜索任务。MapReduce是一种编程模型,用于处理和生成大型数据集,它将...
该框架将任务分解为一系列较小的任务(Map 和 Reduce),并在集群中的多台计算机上并行执行这些任务。 - **应用场景**:适用于大数据分析、搜索引擎索引构建、日志文件分析等场景。 #### 二、MapReduce 工作原理 1...
5. **配置Hadoop连接**:在Eclipse中,选择"Window" -> "Preferences" -> "Hadoop Map/Reduce",配置你的Hadoop集群信息,包括NameNode和JobTracker的地址。 6. **创建Hadoop项目**:使用Eclipse的"File" -> "New" ...
打开Eclipse,依次选择`Window` > `Preferences`,找到`Hadoop Map/Reduce`选项,配置Hadoop的安装目录(即Hadoop的根目录)。如果之前没有安装Hadoop,则需要先下载并安装Hadoop,具体步骤可参考链接:...
然而,利用Hadoop处理数据时,用户必须自己开发Map/Reduce程序,而这种程序处于比较低的层次,不容易掌握且难于维护。为了简化用户的开发工作,Hive应运而生。 Hive是一个建立在Hadoop之上的开源数据仓库工具,它...
【大数据云计算技术系列 hadoop搭建与eclipse开发环境设置】主要涵盖了如何在Windows环境下配置Eclipse开发工具,以便于进行Hadoop项目的...在遇到问题时,应检查配置和权限设置,以确保顺利进行Hadoop应用的开发工作。
MapReduce章节会详述其工作原理,包括Map阶段和Reduce阶段,以及中间键值对的排序和分区过程。通过实际案例,读者可以学习编写MapReduce程序,并了解如何运行和监控作业。同时,还会涉及YARN(Yet Another Resource ...
在Eclipse的"File" -> "New" -> "Project"菜单中,会看到"Hadoop Map/Reduce Project"选项。选择这个选项,按照向导的指示设置项目属性,包括Hadoop版本信息。 接下来,是编写MapReduce程序。Hadoop Eclipse插件...
2. **工作流引擎设计**:针对提出的框架,设计了一个高效的工作流引擎,该引擎能够动态地管理和优化ETL流程,确保每个作业的拆分和组合符合最佳性能标准。 3. **流程级别的优化规则**:根据MapReduce作业的特性及...
3. 重启Eclipse,如果在"Window > Preferences > Hadoop Map/Reduce"中能看到配置选项,说明插件已成功安装。 4. 配置Hadoop集群的连接信息,包括Hadoop的安装路径、NameNode和JobTracker的地址。 需要注意的是,...
Hadoop的MapReduce实现位于`org.apache.hadoop.mapreduce`包下,其中`Mapper`和`Reducer`类是核心接口,分别对应Map和Reduce阶段。开发者需要实现这些接口来定义自己的业务逻辑。 六、工具支持 Hadoop提供了丰富的...
4. 完成安装后,重启Eclipse,即可在"File" -> "New"菜单中看到“Hadoop Map/Reduce Project”选项。 同时,`no.txt`文件可能是用于记录某些说明或注意事项的文本文件,但具体内容需要打开文件查看。在Hadoop开发中...
为了解决这一问题,Hadoop作为一种开源的Map-Reduce实现框架被广泛应用于存储与处理大规模数据集。然而,Map-Reduce编程模型本身较为底层,开发者需要编写大量的自定义程序来实现特定的功能,这不仅增加了开发的难度...
Chukwa包含Agent、Adapter、Collector、Map/Reduce Jobs和HICC等多个组件,协同工作,确保数据的高效流动和处理。 除了离线分析,实时(在线)分析也是大数据处理的重要方面。例如,Apache Storm和Spark Streaming...
Hadoop是一个开源的分布式存储和计算框架,它由Apache软件基金会支持,被广泛应用于大数据的存储、处理和...这些知识点对于理解和使用Hadoop框架至关重要,并且能够帮助技术人员进行实际的Hadoop集群搭建和维护工作。
MapReduce则用于在集群上处理和生成大数据集,它将计算任务分解为Map和Reduce两个阶段处理。除此之外,Hadoop生态系统中还包括了Hive、HBase、Zookeeper等多种技术组件。 对于零基础的学习者,Linux操作系统、虚拟...
阿帕奇·奥兹(Apache Oozie)什么是...Oozie概述Oozie是基于服务器的工作流引擎,专门用于运行具有运行Hadoop Map / Reduce和Pig作业的操作的工作流作业。 Oozie是在Java servlet容器中运行的Java Web应用程序。 出于
4. **Reduce任务执行**:每个Reduce任务从所有Map任务接收相同键的数据,对这些数据进行汇总处理,最后将结果输出。 #### 五、Hadoop生态系统 除了HDFS和MapReduce之外,Hadoop生态系统还包括了其他重要的工具和...
2. 在Eclipse中创建新的Map/Reduce项目,并配置Hadoop的运行环境(如HADOOP_HOME)。 3. 编写MapReduce程序,利用Hadoop提供的API处理数据。 4. 使用Eclipse的“Run As”功能运行MapReduce作业,插件会自动将其提交...