总结以往的工作中遇到的一个问题。

背景：

操作和维护与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
     
       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
     
       {	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原则。

版权声明：本文博客原创文章，博客，未经同意，不得转载。