大家好,又见面了,我是全栈君。
总结以往的工作中遇到的一个问题。
背景: 操作和维护与scribe从apacheserver一再被推到日志记录,所以在这里ETL处理正在进行的重。有根据业务的输出类型是用于多文件夹一个需求。方便挂分区,使用回。 这两种需求都没有问题分开处理,一个mapreduce里完毕,须要一点技巧。
1、map输入数据,经过一系列处理。输出时:
if(ttype.equals("other")){
file = (result.toString().hashCode() & 0x7FFFFFFF)%400;
}else if(ttype.equals("client")){
file = (result.toString().hashCode() & 0x7FFFFFFF)%260;
}else{
file = (result.toString().hashCode()& 0x7FFFFFFF)%60;
}
tp = new TextPair(ttype+"_"+file, result.toString());
context.write(tp, valuet);
valuet是空的,什么都没有。
我这里有三个类型。other,client,wap,分别代表日志来源平台。要按他们分文件夹输出。 result就是整条记录。
file得到的是终于输出文件名称,hash。位操作,取模是为了输出均衡。
map的输出结构<key,value> =(ttype+”_”+file,result.toString()) 这样做的目的是:保证同样的记录得到同样的key,同一时候还要保存类型。partition要按textPair的left,也就是这个key, 保证了后面要写到同一个输出文件的全部记录都到同一个reduce里去。一个reduce能够写多个输出文件。可是一个输出文件不能来自多个reduce,原因非常明了。 这种话大概400+260+60=720个输出文件,每一个文件数据量大概差点儿相同,job的reduce数我这里设置的240,这个数连同取模400,260,60都是依据我的数据量来定的,来尽量避免reduce的数据倾斜。 2、reduce方法去重:
public void reduce(TextPair key, Iterable<Text> values, Context context) throws IOException, InterruptedException
{
rcfileCols = getRcfileCols(key.getSecond().toString().split("\001"));
context.write(key.getFirst(), rcfileCols);
}
不用迭代,对同样的key组。仅仅输出一次。注意这里job用到的比較器,一定不能是FirstComparator,而是整个textpair对的比較。(先比較left。再比較right) 我的程序里输出文件格式是rcfile。 3、多文件夹输出:
job.setOutputFormatClass(WapApacheMutiOutputFormat.class);
public class WapApacheMutiOutputFormat extends RCFileMultipleOutputFormat<Text, BytesRefArrayWritable> {
Random r = new Random();
protected String generateFileNameForKeyValue(Text key, BytesRefArrayWritable value,
Configuration conf) {
String typedir = key.toString().split("_")[0];
return typedir+"/"+key.toString();
}
}
这里的RCFileMultipleOutputFormat是自己继承自FileOutputFormat 自己写的。主要实现了recordWriter。
终于输出去重的,分文件夹的数据文件。
理解的关键,主要是partition key设计。reduce原则。
版权声明:本文博客原创文章,博客,未经同意,不得转载。
发布者:全栈程序员栈长,转载请注明出处:https://javaforall.cn/117394.html原文链接:https://javaforall.cn