java mapreduce（如何通过Java程序提交yarn的MapReduce计算任务）

：暂无数据 2026-08-27 22:50:02 ：0

“java mapreduce”相关信息最新大全有哪些，这是大家都非常关心的，接下来就一起看看java mapreduce（如何通过Java程序提交yarn的MapReduce计算任务）！

本文目录

如何通过Java程序提交yarn的MapReduce计算任务
求mapreduce（JAVA）读取文件然后处理数据的代码
如何将java类对象作为mapreduce中map函数的输入
如何实现mapreduce计算框架以有效实现迭代

如何通过Java程序提交yarn的MapReduce计算任务

　　1、在程序中，我将文件读入格式设定为WholeFileInputFormat，即不对文件进行切分。

　　2、为了控制reduce的处理过程，map的输出键的格式为组合键格式。与常规的《key,value》不同，这里变为了《textpair,value》，TextPair的格式为《key1,key2》。

　　3、为了适应组合键，重新设定了分组函数，即GroupComparator。分组规则为，只要TextPair中的key1相同(不要求key2相同)，则数据被分配到一个reduce容器中。这样，当相同key1的数据进入reduce容器后，key2起到了一个数据标识的作用

求mapreduce（JAVA）读取文件然后处理数据的代码

package test;

import java.io.BufferedReader;
import java.io.File;
import java.io.FileReader;
import java.util.ArrayList;
import java.util.List;

public class Test {
public static void main(String args) {
Test t = new Test();
t.mapreduce(“c:\\a.txt“);
}
public void mapreduce(String filepath){
if(filepath==null||““.equals(filepath)){
System.out.println(“文件名错误！“);
return;
}
List《String》 list = new ArrayList《String》();
File f = new File(filepath);
BufferedReader reader=null;
try {
reader = new BufferedReader(new FileReader(f));
String tempString = null;
// 一次读入一行，直到读入null为文件结束
while ((tempString = reader.readLine()) != null) {
list.add(tempString);
}
reader.close();
} catch (Exception e) {
// TODO Auto-generated catch block
e.printStackTrace();
} finally {
if (reader != null) {
try {
reader.close();
} catch (Exception e1) {
}
}
}
for(int i=0,length=list.size();i《length;i+=3){
String name = list.get(i).split(“:“);
String id = list.get(i+1).split(“:“);
String address = list.get(i+2).split(“:“);
System.out.println(“name:“+name+“,id:“+id+“,address:“+address);
}
}
}

如何将java类对象作为mapreduce中map函数的输入

1.首先介绍一下wordcount 早mapreduce框架中的对应关系
大家都知道 mapreduce 分为 map 和reduce 两个部分，那么在wordcount例子中，很显然对文件word 计数部分为map，对 word 数量累计部分为 reduce；
大家都明白 map接受一个参数，经过map处理后，将处理结果作为reduce的入参分发给reduce，然后在reduce中统计了word 的数量，最终输出到输出结果；
但是初看遇到的问题：
一、map的输入参数是个 Text之类的对象，并不是 file对象
二、reduce中并没有if-else之类的判断语句，来说明这个word 数量加一次，那个word 加一次。那么这个判断到底只是在 map中已经区分了还是在reduce的时候才判断的
三、map过程到底做了什么，reduce过程到底做了什么？为什么它能够做到多个map多个reduce？

一、
1. 怎么将文件参数传递到 job中呢？
在 client 我们调用了FileInputFormat.addInputPath(job, new Path(otherArgs));
实际上 addInputPath 做了以下的事情（将文件路径加载到了conf中）
public static void addInputPath(Job job,
Path path) throws IOException {
Configuration conf = job.getConfiguration();
path = path.getFileSystem(conf).makeQualified(path);
String dirStr = StringUtils.escapeString(path.toString());
String dirs = conf.get(INPUT_DIR);
conf.set(INPUT_DIR, dirs == null ? dirStr : dirs + “,“ + dirStr);
}

我们再来看看 FileInputFormat 是做什么用的， FileInputFormat 实现了 InputFormat 接口，这个接口是hadoop用来接收客户端输入参数的。所有的输入格式都继承于InputFormat，这是一个抽象类，其子类有专门用于读取普通文件的FileInputFormat，用来读取数据库的DBInputFormat等等。

我们会看到在 InputFormat 接口中有getSplits方法，也就是说分片操作实际上实在 map之前就已经做好了
List《InputSplit》getSplits(JobContext job)
Generate the list of files and make them into FileSplits.
具体实现参考 FileInputFormat getSplits 方法：
上面是FileInputFormat的getSplits()方法，它首先得到分片的最小值minSize和最大值maxSize，它们会被用来计算分片大小。可以通过设置mapred.min.split.size和mapred.max.split.size来设置。splits链表用来存储计算得到的输入分片，files则存储作为由listStatus()获取的输入文件列表。然后对于每个输入文件，判断是否可以分割，通过computeSplitSize计算出分片大小splitSize,计算方法是：Math.max(minSize, Math.min(maxSize, blockSize));也就是保证在minSize和maxSize之间，且如果minSize《=blockSize《=maxSize，则设为blockSize。然后我们根据这个splitSize计算出每个文件的inputSplits集合，然后加入分片列表splits中。注意到我们生成InputSplit的时候按上面说的使用文件路径，分片起始位置，分片大小和存放这个文件的hosts列表来创建。最后我们还设置了输入文件数量：mapreduce.input.num.files。

二、计算出来的分片有时怎么传递给 map呢？对于单词数量如何累加？
我们使用了就是InputFormat中的另一个方法createRecordReader() 这个方法：
RecordReader：
RecordReader是用来从一个输入分片中读取一个一个的K -V 对的抽象类，我们可以将其看作是在InputSplit上的迭代器。我们从API接口中可以看到它的一些方法，最主要的方法就是nextKeyvalue()方法，由它获取分片上的下一个K-V 对。

可以看到接口中有：
public abstract boolean nextKeyValue() throws IOException, InterruptedException;
public abstract KEYIN getCurrentKey() throws IOException, InterruptedException;
public abstract VALUEIN getCurrentValue() throws IOException, InterruptedException;
public abstract float getProgress() throws IOException, InterruptedException;
public abstract void close() throws IOException;

FileInputFormat《K,V》
Direct Known Subclasses:
CombineFileInputFormat, KeyValueTextInputFormat, NLineInputFormat, SequenceFileInputFormat, TextInputFormat

对于 wordcount 测试用了 NLineInputFormat和 TextInputFormat 实现类

在 InputFormat 构建一个 RecordReader 出来，然后调用RecordReader initialize 的方法，初始化RecordReader 对象

那么到底 Map是怎么调用的呢？通过前边我们已经将文件分片了，并且将文件分片的内容存放到了RecordReader中，

下面继续看看这些RecordReader是如何被MapReduce框架使用的

终于说道 Map了，我么如果要实现Map 那么一定要继承 Mapper这个类
public abstract class Context
implements MapContext《KEYIN,VALUEIN,KEYOUT,VALUEOUT》 {
}
protected void setup(Context context) throws IOException, InterruptedException
protected void map(KEYIN key, VALUEIN value, Context context) throws IOException,InterruptedException { }
protected void cleanup(Context context ) throws IOException, InterruptedException { }
public void run(Context context) throws IOException, InterruptedException { }

我们写MapReduce程序的时候，我们写的mapper都要继承这个Mapper.class，通常我们会重写map()方法，map()每次接受一个K-V对，然后我们对这个K-V对进行处理，再分发出处理后的数据。我们也可能重写setup()以对这个map task进行一些预处理，比如创建一个List之类的；我们也可能重写cleanup()方法对做一些处理后的工作，当然我们也可能在cleanup()中写出K-V对。举个例子就是：InputSplit的数据是一些整数，然后我们要在mapper中算出它们的和。我们就可以在先设置个sum属性，然后map()函数处理一个K-V对就是将其加到sum上，最后在cleanup()函数中调用context.write(key,value);
最后我们看看Mapper.class中的run()方法，它相当于map task的驱动，我们可以看到run()方法首先调用setup()进行初始操作，然后对每个context.nextKeyValue()获取的K-V对，就调用map()函数进行处理，最后调用cleanup()做最后的处理。事实上，从context.nextKeyValue()就是使用了相应的RecordReader来获取K-V对的。

我们看看Mapper.class中的Context类，它继承与MapContext，使用了一个RecordReader进行构造。下面我们再看这个MapContext。

public MapContextImpl(Configuration conf, TaskAttemptID taskid,
RecordReader《KEYIN,VALUEIN》 reader,
RecordWriter《KEYOUT,VALUEOUT》 writer,
OutputCommitter committer,
StatusReporter reporter,
InputSplit split) {
super(conf, taskid, writer, committer, reporter);
this.reader = reader;
this.split = split;
}

RecordReader 看来是在这里构造出来了，那么是谁调用这个方法，将这个承载着关键数据信息的 RecordReader 传过来了？

我们可以想象这里应该被框架调用的可能性比较大了，那么mapreduce 框架是怎么分别来调用map和reduce呢？
还以为分析完map就完事了，才发现这里仅仅是做了mapreduce 框架调用前的一些准备工作，

还是继续分析下 mapreduce 框架调用吧：

1.在 job提交任务之后首先由jobtrack 分发任务，

在任务分发完成之后，执行 task的时候，这时调用了 maptask 中的 runNewMapper

在这个方法中调用了 MapContextImpl，至此这个map 和框架就可以联系起来了。

如何实现mapreduce计算框架以有效实现迭代

　　MapReduce从出现以来，已经成为Apache Hadoop计算范式的扛鼎之作。它对于符合其设计的各项工作堪称完美：大规模日志处理，ETL批处理操作等。

　　随着Hadoop使用范围的不断扩大，人们已经清楚知道MapReduce不是所有计算的最佳框架。Hadoop 2将资源管理器YARN作为自己的顶级组件，为其他计算引擎的接入提供了可能性。如Impala等非MapReduce架构的引入，使平台具备了支持交互式SQL的能力。

　　今天，Apache Spark是另一种这样的替代，并且被称为是超越MapReduce的通用计算范例。也许您会好奇：MapReduce一直以来已经这么有用了，怎么能突然被取代？毕竟，还有很多ETL这样的工作需要在Hadoop上进行，即使该平台目前也已经拥有其他实时功能。

　　值得庆幸的是，在Spark上重新实现MapReduce一样的计算是完全可能的。它们可以被更简单的维护，而且在某些情况下更快速，这要归功于Spark优化了刷写数据到磁盘的过程。Spark重新实现MapReduce编程范式不过是回归本源。Spark模仿了Scala的函数式编程风格和API。而MapReduce的想法来自于函数式编程语言LISP。

　　尽管Spark的主要抽象是RDD（弹性分布式数据集），实现了Map，reduce等操作，但这些都不是Hadoop的Mapper或Reducer API的直接模拟。这些转变也往往成为开发者从Mapper和Reducer类平行迁移到Spark的绊脚石。

　　与Scala或Spark中经典函数语言实现的map和reduce函数相比，原有Hadoop提供的Mapper和Reducer API 更灵活也更复杂。这些区别对于习惯了MapReduce的开发者而言也许并不明显，下列行为是针对Hadoop的实现而不是MapReduce的抽象概念：
　　· Mapper和Reducer总是使用键值对作为输入输出。
　　· 每个Reducer按照Key对Value进行reduce。
　　· 每个Mapper和Reducer对于每组输入可能产生0个，1个或多个键值对。
　　· Mapper和Reducer可能产生任意的keys和values，而不局限于输入的子集和变换。
　　Mapper和Reducer对象的生命周期可能横跨多个map和reduce操作。它们支持setup和cleanup方法，在批量记录处理开始之前和结束之后被调用。

　　本文将简要展示怎样在Spark中重现以上过程，您将发现不需要逐字翻译Mapper和Reducer！

　　作为元组的键值对
　　假定我们需要计算大文本中每一行的长度，并且报告每个长度的行数。在HadoopMapReduce中，我们首先使用一个Mapper，生成为以行的长度作为key，1作为value的键值对。
　　public class LineLengthMapper extends
　　Mapper《LongWritable, Text, IntWritable, IntWritable》 {
　　@Override
　　protected void map(LongWritable lineNumber, Text line, Context context)
　　throws IOException, InterruptedException {
　　context.write(new IntWritable(line.getLength()), new IntWritable(1));
　　}
　　}

　　值得注意的是Mappers和Reducers只对键值对进行操作。所以由TextInputFormat提供输入给LineLengthMapper，实际上也是以文本中位置为key（很少这么用，但是总是需要有东西作为Key），文本行为值的键值对。

　　与之对应的Spark实现：

　　lines.map(line =》 (line.length, 1))

　　Spark中，输入只是String构成的RDD，而不是key-value键值对。Spark中对key-value键值对的表示是一个Scala的元组，用(A，B)这样的语法来创建。上面的map操作的结果是(Int，Int)元组的RDD。当一个RDD包含很多元组，它获得了多个方法，如reduceByKey，这对再现MapReduce行为将是至关重要的。

　　Reduce
　　reduce()与reduceBykey()
　　统计行的长度的键值对，需要在Reducer中对每种长度作为key，计算其行数的总和作为value。
　　public class LineLengthReducer extends
　　Reducer《IntWritable, IntWritable, IntWritable, IntWritable》 {
　　@Override
　　protected void reduce(IntWritable length, Iterable《IntWritable》 counts,
　　Context context) throws IOException, InterruptedException {
　　int sum = 0;
　　for (IntWritable count : counts) {
　　sum += count.get();
　　}
　　context.write(length, new IntWritable(sum));
　　}
　　}

　　Spark中与上述Mapper，Reducer对应的实现只要一行代码：
　　val lengthCounts = lines.map(line =》 (line.length, 1)).reduceByKey(_ + _)

　　Spark的RDD API有个reduce方法，但是它会将所有key-value键值对reduce为单个value。这并不是Hadoop MapReduce的行为，Spark中与之对应的是ReduceByKey。

　　另外，Reducer的Reduce方法接收多值流，并产生0，1或多个结果。而reduceByKey，它接受的是一个将两个值转化为一个值的函数，在这里，就是把两个数字映射到它们的和的简单加法函数。此关联函数可以被调用者用来reduce多个值到一个值。与Reducer方法相比，他是一个根据Key来Reduce Value的更简单而更精确的API。
　　Mapper
　　map() 与 flatMap()
　　现在，考虑一个统计以大写字母开头的单词的个数的算法。对于每行输入文本，Mapper可能产生0个，1个或多个键值对。
　　public class CountUppercaseMapper extends
　　Mapper《LongWritable, Text, Text, IntWritable》 {
　　@Override
　　protected void map(LongWritable lineNumber, Text line, Context context)
　　throws IOException, InterruptedException {
　　for (String word : line.toString().split(“ “)) {
　　if (Character.isUpperCase(word.charAt(0))) {
　　context.write(new Text(word), new IntWritable(1));
　　}
　　}
　　}
　　}

　　Spark对应的写法：
　　lines.flatMap(
　　_.split(“ “).filter(word =》 Character.isUpperCase(word(0))).map(word =》 (word,1))
　　)
　　简单的Spark map函数不适用于这种场景，因为map对于每个输入只能产生单个输出，但这个例子中一行需要产生多个输出。所以，和MapperAPI支持的相比，Spark的map函数语义更简单，应用范围更窄。

　　Spark的解决方案是首先将每行映射为一组输出值，这组值可能为空值或多值。随后会通过flatMap函数被扁平化。数组中的词会被过滤并被转化为函数中的元组。这个例子中，真正模仿Mapper行为的是flatMap，而不是map。
　　groupByKey()
　　写一个统计次数的reducer是简单的，在Spark中，reduceByKey可以被用来统计每个单词的总数。比如出于某种原因要求输出文件中每个单词都要显示为大写字母和其数量，在MapReduce中，实现如下：
　　public class CountUppercaseReducer extends
　　Reducer《Text, IntWritable, Text, IntWritable》 {
　　@Override
　　protected void reduce(Text word, Iterable《IntWritable》 counts, Context context)
　　throws IOException, InterruptedException {
　　int sum = 0;
　　for (IntWritable count : counts) {
　　sum += count.get();
　　}
　　context
　　.write(new Text(word.toString().toUpperCase()), new IntWritable(sum));
　　}
　　}

　　但是redeceByKey不能单独在Spark中工作，因为他保留了原来的key。为了在Spark中模拟，我们需要一些更像Reducer API的操作。我们知道Reducer的reduce方法接受一个key和一组值，然后完成一组转换。groupByKey和一个连续的map操作能够达到这样的目标:
　　groupByKey().map { case (word,ones) =》 (word.toUpperCase, ones.sum) }
　　groupByKey只是将某一个key的所有值收集在一起，并且不提供reduce功能。以此为基础，任何转换都可以作用在key和一系列值上。此处，将key转变为大写字母，将values直接求和。

　　setup()和cleanup()

　　在MapReduce中，Mapper和Reducer可以声明一个setup方法，在处理输入之前执行，来进行分配数据库连接等昂贵资源，同时可以用cleanup函数可以释放资源。
　　public class SetupCleanupMapper extends
　　Mapper《LongWritable, Text, Text, IntWritable》 {
　　private Connection dbConnection;
　　@Override
　　protected void setup(Context context) {
　　dbConnection = ...;
　　}
　　...
　　@Override
　　protected void cleanup(Context context) {
　　dbConnection.close();
　　}
　　}

　　Spark中的map和flatMap方法每次只能在一个input上操作，而且没有提供在转换大批值前后执行代码的方法，看起来，似乎可以直接将setup和cleanup代码放在Sparkmap函数调用之前和之后：
　　val dbConnection = ...
　　lines.map(... dbConnection.createStatement(...) ...)
　　dbConnection.close() // Wrong!

　　然而这种方法却不可行，原因在于：
　　· 它将对象dbConnection放在map函数的闭包中，这需要他是可序列化的（比如，通过java.io.Serializable实现）。而数据库连接这种对象一般不能被序列化。
　　· map是一种转换，而不是操作，并且拖延执行。连接对象不能被及时关闭。
　　· 即便如此，它也只能关闭driver上的连接，而不是释放被序列化拷贝版本分配的资源连接。

　　事实上，map和flatMap都不是Spark中Mapper的最接近的对应函数，Spark中Mapper的最接近的对应函数是十分重要的mapPartitions()方法，这个方法能够不仅完成单值对单值的映射，也能完成一组值对另一组值的映射，很像一个批映射（bulkmap）方法。这意味着mapPartitions()方法能够在开始时从本地分配资源，并在批映射结束时释放资源。

　　添加setup方法是简单的，添加cleanup会更困难，这是由于检测转换完成仍然是困难的。例如，这样是能工作的：

　　lines.mapPartitions { valueIterator =》
　　val dbConnection = ... // OK
　　val transformedIterator = valueIterator.map(... dbConnection ...)
　　dbConnection.close() // Still wrong! May not have evaluated iterator
　　transformedIterator
　　}

　　一个完整的范式应该看起来类似于：
　　lines.mapPartitions { valueIterator =》
　　if (valueIterator.isEmpty) {
　　Iterator[...]()
　　} else {
　　val dbConnection = ...
　　valueIterator.map { item =》
　　val transformedItem = ...
　　if (!valueIterator.hasNext) {
　　dbConnection.close()
　　}
　　transformedItem
　　}
　　}
　　}

　　虽然后者代码翻译注定不如前者优雅，但它确实能够完成工作。

　　flatMapPartitions方法并不存在，然而，可以通过调用mapPartitions，后面跟一个flatMap(a= 》 a)的调用达到同样效果。

　　带有setup和cleanup的Reducer对应只需仿照上述代码使用groupByKey后面跟一个mapPartition函数。

　　别急，等一下，还有更多
　　MapReduce的开发者会指出，还有更多的还没有被提及的API：
　　· MapReduce支持一种特殊类型的Reducer，也称为Combiner，可以从Mapper中减少洗牌(shuffled)数据大小。
　　· 它还支持同通过Partitioner实现的自定义分区，和通过分组Comparator实现的自定义分组。
　　· Context对象授予Counter API的访问权限以及它的累积统计。
　　· Reducer在其生命周期内一直能看到已排序好的key 。
　　· MapReduce有自己的Writable序列化方案。
　　· Mapper和Reducer可以一次发射多组输出。
　　· MapReduce有几十个调优参数。

　　有很多方法可以在Spark中实现这些方案，使用类似Accumulator的API，类似groupBy和在不同的这些方法中加入partitioner参数的方法，Java或Kryo序列化，缓存和更多。由于篇幅限制，在这篇文章中就不再累赘介绍了。

　　需要指出的是，MapReduce的概念仍然有用。只不过现在有了一个更强大的实现，并利用函数式语言，更好地匹配其功能性。理解Spark RDD API和原来的Mapper和ReducerAPI之间的差异，可以帮助开发者更好地理解所有这些函数的工作原理，以及理解如何利用Spark发挥其优势。