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

如何将java类对象作为mapreduce中map函数的输入?,第1张

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 我们调用了FileInputFormataddInputPath(job, new Path(otherArgs[0]));
实际上 addInputPath 做了以下的事情(将文件路径加载到了conf中)
public static void addInputPath(Job job,
Path path) throws IOException {
Configuration conf = jobgetConfiguration();
path = pathgetFileSystem(conf)makeQualified(path);
String dirStr = StringUtilsescapeString(pathtoString());
String dirs = confget(INPUT_DIR);
confset(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,它们会被用来计算分片大小。可以通过设置mapredminsplitsize和mapredmaxsplitsize来设置。splits链表用来存储计算得到的输入分片,files则存储作为由listStatus()获取的输入文件列表。然后对于每个输入文件,判断是否可以分割,通过computeSplitSize计算出分片大小splitSize,计算方法是:Mathmax(minSize, Mathmin(maxSize, blockSize));也就是保证在minSize和maxSize之间,且如果minSize<=blockSize<=maxSize,则设为blockSize。然后我们根据这个splitSize计算出每个文件的inputSplits集合,然后加入分片列表splits中。注意到我们生成InputSplit的时候按上面说的使用文件路径,分片起始位置,分片大小和存放这个文件的hosts列表来创建。最后我们还设置了输入文件数量:mapreduceinputnumfiles。
二、计算出来的分片有时怎么传递给 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都要继承这个Mapperclass,通常我们会重写map()方法,map()每次接受一个K-V对,然后我们对这个K-V对进行处理,再分发出处理后的数据。我们也可能重写setup()以对这个map task进行一些预处理,比如创建一个List之类的;我们也可能重写cleanup()方法对做一些处理后的工作,当然我们也可能在cleanup()中写出K-V对。举个例子就是:InputSplit的数据是一些整数,然后我们要在mapper中算出它们的和。我们就可以在先设置个sum属性,然后map()函数处理一个K-V对就是将其加到sum上,最后在cleanup()函数中调用contextwrite(key,value);
最后我们看看Mapperclass中的run()方法,它相当于map task的驱动,我们可以看到run()方法首先调用setup()进行初始 *** 作,然后对每个contextnextKeyValue()获取的K-V对,就调用map()函数进行处理,最后调用cleanup()做最后的处理。事实上,从contextnextKeyValue()就是使用了相应的RecordReader来获取K-V对的。
我们看看Mapperclass中的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);
thisreader = reader;
thissplit = split;
}
RecordReader 看来是在这里构造出来了, 那么 是谁调用这个方法,将这个承载着关键数据信息的 RecordReader 传过来了 ?
我们可以想象 这里 应该被框架调用的可能性比较大了,那么mapreduce 框架是怎么分别来调用map和reduce呢?
还以为分析完map就完事了,才发现这里仅仅是做了mapreduce 框架调用前的一些准备工作,
还是继续分析 下 mapreduce 框架调用吧:
1在 job提交 任务之后 首先由jobtrack 分发任务,
在 任务分发完成之后 ,执行 task的时候,这时 调用了 maptask 中的 runNewMapper
在这个方法中调用了 MapContextImpl, 至此 这个map 和框架就可以联系起来了。

list是个接口,arrayList是个实体类。。。
你要是有需求这样转化就相当于要将arrayList转化为List
如果真出现这种情况就说明你声明的就出了问题
要转化也可以,只不过也没必要
转化的方法:重新定义一个Map对象,将其泛型定义为<String,List<String>>
然后给此Map赋予以前的key
例子:
//模拟已存在的Map<String, ArrayList<String>>类型数据
//其key为"number",value为1-100
ArrayList<String> arList = new ArrayList<String>();
Map<String, ArrayList<String>> map = new HashMap<String, ArrayList<String>>();
for (int i = 0; i < 100; i++) {
arListadd((1+i)+"");
}
mapput("number", arList);
//模拟数据结
List<String> list = new ArrayList<String>();
listaddAll(arList);
Map<String,List<String>> map2 = new HashMap<String,List<String>>();
map2put("number", list);
如果key为多个值,请用iterator进行全部遍历。
另外,出方法的时候最好将原有的引用置空,以方便GC可以在第一时间回收

其实可以多利用String的split方法。
换行符的话可以认为是\
和\\r,然后把字符串的每一行拿出来,再对每一行split,得到所需要的键和值。
我写了个例子,功能可能不够完善,不过大问题是没有了,供你参考:
public class Str2MapImpl implements Str2Map {

private String status;

public Map<String, String> getStatus() {

if(status==null)
return null;

Map<String,String> statusMap = new HashMap<String,String>();

String delim = "\
|\\r";

String[] rows = statussplit(delim);

for(int i=0;i<rowslength;i++){
String row = rows[i];
if(rowequals(""))
continue;
String[] keyAndValue = rowsplit("=");
if(keyAndValuelength<2)
continue;
String key = keyAndValue[0];
String value = keyAndValue[1];
if(keyequals("")||valueequals(""))
continue;
statusMapput(key, value);

}

return statusMap;

}

public void setStatusStr(String status_) {

status = status_;

}

public static void main(String args[]){
//测试用例

Str2MapImpl s2m= new Str2MapImpl();

String str = "a=1\
b=2\\rc=3";

s2msetStatusStr(str);

Map<String,String> status = s2mgetStatus();

Systemoutprintln(status);

}
}


欢迎分享,转载请注明来源:内存溢出

原文地址: http://outofmemory.cn/yw/13335945.html

(0)
打赏 微信扫一扫 微信扫一扫 支付宝扫一扫 支付宝扫一扫
上一篇 2023-07-17
下一篇 2023-07-17

发表评论

登录后才能评论

评论列表(0条)

保存