`
brandNewUser
  • 浏览: 455437 次
  • 性别: Icon_minigender_1
  • 来自: 北京
社区版块
存档分类
最新评论

Hadoop Map/Reduce的工作流

阅读更多

问题描述

我们的数据分析平台是单一的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状态的作业等。

ChainMapper/ChainReduce

ChainMapper/ChainReducer主要是为了解决线性链式Mapper而提出的,在Map或Reduce阶段存在多个Mapper,像多个Linux管道一样,前一个Mapper的输出结果直接重定向到下一个Mapper的输入,形成一个流水线,最后的Mapper或Reducer才会将结果写到HDFS上。对于任意一个MapReduce作业,Map和Reduce阶段可以由无限个Mapper,但只能有一个Reducer。

Hadoop MapReduce有一个约定,函数OutputCollector.collect(key, value)执行期间不能改变key和value的值,这是因为某个map/reduce调用该方法之后,可能后续继续再次使用key和value的值,如果被改变,可能会造成潜在的错误。

ChainMapper/Reducer实现的关键技术点就是修改Mapper和Reducer的输出流,将本来要写入文件的输出结果重定向到另外一个Mapper中。尽管链式作业在Map和Reduce阶段添加了多个Mapper,但仍然只是一个MapReduce作业,因而只能有一个与之对应的JobConf对象。

ChainMapper中实现的map函数大概如下:

public void map(Object key, Object value, OutputCollector output, Reporter reporter) throws IOException{
     Mapper mapper = chain.getFirstMap();
     if(mapper != null){
          mapper.map(key, value, chain.getMapperCollector(0, output, reporter), reporter);
     }
}

 

chain.getMapperCollector返回一个OutputCollector实现,即ChainOutputCollector,collector方法大概如下:

public void collect(K key, V value) throws IOException{
     if(nextMapperIndex < mappers.size()){     
          //调用下一个Mapper,直到没有mapper
          nextMapper.map(key, value, new ChainOutputCollector(nextMapperIndex, nextKeySerialization, nextValueSerialization, output, reporter));
     } else {
          //如果是最后一个Mapper,直接调用真正的Collector
          output.collect(key, value);
     }
}

 

 在使用ChainMapper/ChainReducer时需要注意一个问题:就是其中参数byValue的选择,究竟是该传值还是传递引用。因为在Hadoop编程中需要处理的数据量比较大,经常使用复用同一个对象的情况,普通的Mapper/Reducer程序由于不会执行链式处理,在其他的JVM中来重建Map输出的对象,而Chain API中需要管道一样的操作来进行下一步处理,Mapper.map()函数调用完outputCollector.collect(key, value)之后,可能再次使用key和value的值,才导致这个问题的发生。

 

个人总觉得虽然重用引用的方式虽然可以节省一定的内存,但是不重用引用也仅仅会对Minor GC造成一定的压力,如果严格控制生成的new对象Key,Value的生命周期的话。

 

正是为了防止OutputCollector直接对key/value进行修改,ChainMapper允许用户指定key/value的传递方式,如果编写的程序确定key/value执行期间不会被重用以修改(如果是不可变对象最好),则可以选择按照引用来进行传递,否则按值传递。需要注意的是,引用传递可以避免对象的深层拷贝,提高处理效率,但需要编程时做出key/value不能修改的保证。

 

 

  • 大小: 68.3 KB
  • 大小: 35.6 KB
  • 大小: 50.6 KB
分享到:
评论

相关推荐

    远程调用执行Hadoop Map/Reduce

    7. **工具集成**:有许多开源工具可以帮助我们远程提交和管理Hadoop作业,如Hadoop命令行工具、Hadoop的Web UI、Apache Oozie工作流管理系统等。这些工具提供了方便的接口,使开发者能便捷地与集群交互。 8. **安全...

    在solr文献检索中用map/reduce

    标题中的“在solr文献检索中用map/reduce”指的是使用Apache Solr,一个流行的开源搜索引擎,结合Hadoop的MapReduce框架来处理大规模的分布式搜索任务。MapReduce是一种编程模型,用于处理和生成大型数据集,它将...

    Hadoop Map Reduce教程

    该框架将任务分解为一系列较小的任务(Map 和 Reduce),并在集群中的多台计算机上并行执行这些任务。 - **应用场景**:适用于大数据分析、搜索引擎索引构建、日志文件分析等场景。 #### 二、MapReduce 工作原理 1...

    安装hadoop使用Eclipse的插件

    5. **配置Hadoop连接**:在Eclipse中,选择"Window" -&gt; "Preferences" -&gt; "Hadoop Map/Reduce",配置你的Hadoop集群信息,包括NameNode和JobTracker的地址。 6. **创建Hadoop项目**:使用Eclipse的"File" -&gt; "New" ...

    基于eclipse的hadoop应用开发

    打开Eclipse,依次选择`Window` &gt; `Preferences`,找到`Hadoop Map/Reduce`选项,配置Hadoop的安装目录(即Hadoop的根目录)。如果之前没有安装Hadoop,则需要先下载并安装Hadoop,具体步骤可参考链接:...

    基于Hadoop/Hive的web日志分析系统的设计

    然而,利用Hadoop处理数据时,用户必须自己开发Map/Reduce程序,而这种程序处于比较低的层次,不容易掌握且难于维护。为了简化用户的开发工作,Hive应运而生。 Hive是一个建立在Hadoop之上的开源数据仓库工具,它...

    大数据云计算技术系列 hadoop搭建与eclipse开发环境设置-已验证通过(共13页).pdf

    【大数据云计算技术系列 hadoop搭建与eclipse开发环境设置】主要涵盖了如何在Windows环境下配置Eclipse开发工具,以便于进行Hadoop项目的...在遇到问题时,应检查配置和权限设置,以确保顺利进行Hadoop应用的开发工作。

    hadoop中文实战

    MapReduce章节会详述其工作原理,包括Map阶段和Reduce阶段,以及中间键值对的排序和分区过程。通过实际案例,读者可以学习编写MapReduce程序,并了解如何运行和监控作业。同时,还会涉及YARN(Yet Another Resource ...

    hadoop eclipse插件

    在Eclipse的"File" -&gt; "New" -&gt; "Project"菜单中,会看到"Hadoop Map/Reduce Project"选项。选择这个选项,按照向导的指示设置项目属性,包括Hadoop版本信息。 接下来,是编写MapReduce程序。Hadoop Eclipse插件...

    基于MapReduce作业拆分组合机制的并行ETL组件实现.pdf

    2. **工作流引擎设计**:针对提出的框架,设计了一个高效的工作流引擎,该引擎能够动态地管理和优化ETL流程,确保每个作业的拆分和组合符合最佳性能标准。 3. **流程级别的优化规则**:根据MapReduce作业的特性及...

    hadoop-eclipse-plugin-2.7.3.jar

    3. 重启Eclipse,如果在"Window &gt; Preferences &gt; Hadoop Map/Reduce"中能看到配置选项,说明插件已成功安装。 4. 配置Hadoop集群的连接信息,包括Hadoop的安装路径、NameNode和JobTracker的地址。 需要注意的是,...

    a java map reduce framework

    Hadoop的MapReduce实现位于`org.apache.hadoop.mapreduce`包下,其中`Mapper`和`Reducer`类是核心接口,分别对应Map和Reduce阶段。开发者需要实现这些接口来定义自己的业务逻辑。 六、工具支持 Hadoop提供了丰富的...

    hadoop-eclipse-plugin-2.7.1.jar

    4. 完成安装后,重启Eclipse,即可在"File" -&gt; "New"菜单中看到“Hadoop Map/Reduce Project”选项。 同时,`no.txt`文件可能是用于记录某些说明或注意事项的文本文件,但具体内容需要打开文件查看。在Hadoop开发中...

    Hive - A Warehousing Solution Over a Map-Reduce.pdf

    为了解决这一问题,Hadoop作为一种开源的Map-Reduce实现框架被广泛应用于存储与处理大规模数据集。然而,Map-Reduce编程模型本身较为底层,开发者需要编写大量的自定义程序来实现特定的功能,这不仅增加了开发的难度...

    海量数据处理平台体系架构分析[参照].pdf

    Chukwa包含Agent、Adapter、Collector、Map/Reduce Jobs和HICC等多个组件,协同工作,确保数据的高效流动和处理。 除了离线分析,实时(在线)分析也是大数据处理的重要方面。例如,Apache Storm和Spark Streaming...

    Hadoop-MindMap 技术全解

    Hadoop是一个开源的分布式存储和计算框架,它由Apache软件基金会支持,被广泛应用于大数据的存储、处理和...这些知识点对于理解和使用Hadoop框架至关重要,并且能够帮助技术人员进行实际的Hadoop集群搭建和维护工作。

    零基础学习Hadoop

    MapReduce则用于在集群上处理和生成大数据集,它将计算任务分解为Map和Reduce两个阶段处理。除此之外,Hadoop生态系统中还包括了Hive、HBase、Zookeeper等多种技术组件。 对于零基础的学习者,Linux操作系统、虚拟...

    oozie:Apache Oozie的镜子

    阿帕奇·奥兹(Apache Oozie)什么是...Oozie概述Oozie是基于服务器的工作流引擎,专门用于运行具有运行Hadoop Map / Reduce和Pig作业的操作的工作流作业。 Oozie是在Java servlet容器中运行的Java Web应用程序。 出于

    hadoop手册-分布式框架介绍

    4. **Reduce任务执行**:每个Reduce任务从所有Map任务接收相同键的数据,对这些数据进行汇总处理,最后将结果输出。 #### 五、Hadoop生态系统 除了HDFS和MapReduce之外,Hadoop生态系统还包括了其他重要的工具和...

    hadoop2.6.0源码和eclipse插件

    2. 在Eclipse中创建新的Map/Reduce项目,并配置Hadoop的运行环境(如HADOOP_HOME)。 3. 编写MapReduce程序,利用Hadoop提供的API处理数据。 4. 使用Eclipse的“Run As”功能运行MapReduce作业,插件会自动将其提交...

Global site tag (gtag.js) - Google Analytics