离线工程代码学习总结

1.背景概述：

离线筛选工程，这个工程的代码结构和设计思想还是很优秀的，但离线筛选这个工程一直没有参与生产。也一直没有维护用起来，晒一下其中精华的部分。

2.技术概述

2.1 部署结构：



(1) 开发好的代码通过FTP上传到FTP服务器。

(2) gateway机器从ftp上下载代码并部署web应用。

(3) 调度job：

I. 调度gateway凌晨定时向云梯提交构建宽表的任务。

II.调度gateway实时的向云梯提交离线筛选任务，并且监听job执行进度完成hbase回流。

(4) 云梯任务回流Hbase。

(5) 淘营销读取Hbase的数据，展现。

(6) 管理后台提交离线筛选任务。

(7) 调度gateway监控是否有需要消费的离线筛选任务，并向云梯提交离线筛选任务。

描述：目前现有交接过来的工程实现的反向招商的思路是，以商品AA表为基础，通过凌晨定时任务将需要做离线筛选的的指标项按照商品维度拼装到商品记录行，形成宽表。然后按照淘营销指定的活动资质的维度对商品进行实时的筛选和回流。

2.2 关键代码模块：



（1）代码程序的主入口是MapredServer，这个任务主要启动四个线程任务 MapredTaskThread，ReadQualificationThread，DataFlowToHbaseThread，QuartzScheduler。

（2）QuartzScheduler这个任务主要定时的从云梯上读取商品、用户、交易的信息，然后以商品维度在云梯上生成宽表，作为离线筛选任务的数据源。

（3）ReadQualificationThread 这个线程任务通过 QualifiedServer（资质筛选任务管理引擎）和qc中心产生交互，读取需要执行的离线筛选任务放入到离线筛选任务队列（queue）。

（4）MapredTaskThread 这个线程任务消耗离线筛选任务队列里的离线筛选任务，将离线筛选任务封装成动态的封装成 MapReduce任务通过gateway上部署的hadoop client提交的云梯上执行。监控执行结果，执行成功后将任务加入到数据回流队列（dfqueue）。

（5）DataFlowToHbaseThread 这个线程任务，消耗离线筛选任务队列里的任务，将需要回流的数据通过Hbase client API将数据写入Hbase集群。 （6)离线筛选任务主要路径描述完毕。

2.3 代码技术细节：

性能优化点: 重写了MR的输出的OutputFormat和OutputKey。

背景： 这个离线筛选工程和和淘营销、资质中心结合比较紧密的，一个离线筛选任务对应一个淘营销的活动。而淘营销活动数量是很多的，如果将每个离线筛选任务都封装成一个MR任务提交到云梯上，其实由于任务数量太多，任务计算的性能是比较差的。

优化： 将多个离线筛选任务封装成一个MR计算任务，然后再输出时按照离线筛选任务的ID分目录输出。



重写的代码继承结构：



简述：其中没有用类全限定名的是资质重写的类，一个是QualiOutputFormat，里面重写了Reduce数据的输出路径，核心代码如下：

@Override
public RecordWriter<QualiWritable, Writable> getRecordWriter(
FileSystem fs, JobConf jobConf, String name, Progressable progress)
throws IOException {
final FileSystem myFS = fs;
final String myName = name;
final JobConf myJob = jobConf;
final Progressable myProgressable = progress;
final Path outDir = FileOutputFormat.getOutputPath(jobConf);

return new RecordWriter<QualiWritable, Writable>() {
TreeMap<String, RecordWriter<Writable, Writable>> recordWriters = new TreeMap<String, RecordWriter<Writable, Writable>>();

public void write(QualiWritable key, Writable value)
throws IOException {
//定义了Reduce输出数据的路径,可以按照离线任务分目录输出
String keyBasedPath = key.getName() + "/" + myName;
RecordWriter<Writable, Writable> rw = this.recordWriters
.get(keyBasedPath);
if (rw == null) {
Path path = new Path(outDir, keyBasedPath);
rw = outputFormat.getRecordWriter(myFS, myJob,
path.toString(), myProgressable);
this.recordWriters.put(keyBasedPath, rw);
if (!myFS.exists(path)) myFS.mkdirs(path);
}
rw.write(key.getRawWritable(), value);

}

因为多个离线筛选任务合并成了一个云梯任务，重写的这个Write方法实现了输出数据按照离线筛选任务的ID分目录输出。

另一个是QualiWritable，这个类重写了OutputKey的排序方式，定义输出数据的顺序，将同一个离线筛选ID的数据聚集在一起，方便输出。

设置这两个重写类的JOB代码片段如下：

jobConf.setOutputFormat(QualiOutputFormat.class);
FileOutputFormat.setOutputPath(jobConf, parentOutDir);

jobConf.setMapOutputKeyClass(Text.class);
jobConf.setMapOutputValueClass(Text.class);
/**
* 实现了QualiWritable，优化outputkey
*/
jobConf.setOutputKeyClass(QualiWritable.class);
jobConf.setOutputValueClass(NullWritable.class);
if (taskDo.getFilterType() == FilterType.auction) {
jobConf.setMapperClass(AuctionExecMapper.class);
} else if (taskDo.getFilterType() == FilterType.shop) {
jobConf.setMapperClass(ShopExecMapper.class);
}
jobConf.setReducerClass(ExecReduce.class);
jobClient = new JobClient(jobConf);
runingJob = jobClient.submitJob(jobConf);

总结：这个工程对于MR输出类的重写，提升了离线筛选任务总体的执行效率，可以说是这个工程的亮点，以后有相似场景可以借鉴。