统计电视机顶盒中无效用户数据，并以压缩格式输出

package com.buaa;

import java.util.ArrayList;
import java.util.List;

import org.apache.commons.lang.StringUtils;
import org.jsoup.Jsoup;
import org.jsoup.nodes.Document;
import org.jsoup.nodes.Element;
import org.jsoup.select.Elements;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/** 
* @ProjectName CountSetTopBoxUserData
* @PackageName com.buaa
* @ClassName ParseTVData
* @Description 解析数据
* @Author 刘吉超
* @Date 2016-05-28 16:15:08
*/
public class ParseTVData {
    private static Logger logger = LoggerFactory.getLogger(ParseTVData.class);
    
    /**
     * 使用 Jsoup 工具，解析输入数据
     * 
     * @param text
     * @return List
     */
    public static List<String> transData(String text) {
        List<String> list = new ArrayList<String>();
        
        try {
            // jsoup解析数据
            Document doc = Jsoup.parse(text);
            Elements content = doc.getElementsByTag("WIC");
            
            // 机顶盒号
            String stbNum = content.get(0).attr("stbNum");
            if(StringUtils.isEmpty(stbNum)){
                return list;
            }
            
            // 日期
            String date = content.get(0).attr("date");
            
            Elements els = doc.getElementsByTag("A");
            if (els.isEmpty()) {
                return list;
            }
            
            for (Element el : els) {
                // 结束时间
                String e = el.attr("e");
                // 开始时间
                String s = el.attr("s");
                // 频道名称
                String sn = el.attr("sn");
                
                StringBuilder rec = new StringBuilder().append(stbNum).append("@").append(date).append("@").append(sn).append("@").append(s).append("@").append(e);
                
                list.add(rec.toString());
            }
        } catch (Exception e) {
            logger.error("", e);
            return list;
        }
        return list;
    }
}

package com.buaa;

import java.util.List;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.conf.Configured;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.io.compress.GzipCodec;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
import org.apache.hadoop.util.Tool;
import org.apache.hadoop.util.ToolRunner;

/** 
* @ProjectName CountSetTopBoxUserData
* @PackageName com.buaa
* @ClassName CountUserData
* @Description 统计电视机顶盒中无效用户数据，并以压缩格式输出
* @Author 刘吉超
* @Date 2016-05-28 16:11:12
*/
public class CountUserData extends Configured implements Tool {
    // 定义枚举对象
    public static enum LOG_PROCESSOR_COUNTER {
        BAD_RECORDS
    };
    
    /**
     * 解析数据，统计无效数据，并输出有效数据
     */
    public static class CounterAndCompressionMapper extends Mapper<LongWritable, Text, Text, Text> {
        protected void map(LongWritable key, Text value, Context context) throws java.io.IOException, InterruptedException {
            // 解析每条机顶盒记录，返回list集合
            List<String> list = ParseTVData.transData(value.toString());
            
            // 无效记录
            if (list.isEmpty()) {
                // 动态自定义计数器
                context.getCounter("ErrorRecordCounter", "ERROR_Record_TVData").increment(1);
                // 枚举声明计数器
                context.getCounter(LOG_PROCESSOR_COUNTER.BAD_RECORDS).increment(1);
            } else {
                for (String validateRecord : list) {
                    //输出解析数据
                    context.write(new Text(validateRecord), new Text(""));
                }
            }

        }
    }
    
    @SuppressWarnings("deprecation")
    @Override
    public int run(String[] args) throws Exception {
        // 读取配置文件
        Configuration conf = new Configuration();
        
        // 如果输出目录存在，则删除 
        Path mypath = new Path(args[1]);
        FileSystem hdfs = mypath.getFileSystem(conf);
        if (hdfs.isDirectory(mypath)) {
            // 删除已经存在的文件路径
            hdfs.delete(mypath, true);
        }
        
        // 新建一个任务
        Job job = new Job(conf, "CountUserData");
        // 设置主类
        job.setJarByClass(CountUserData.class);
        
        // Mapper
        job.setMapperClass(CounterAndCompressionMapper.class);
        
        // 输出key类型
        job.setOutputKeyClass(Text.class);
        // 输出value类型
        job.setOutputValueClass(Text.class);
        
        // 输入路径
        FileInputFormat.addInputPath(job, new Path(args[0]));
        // 输出路径
        FileOutputFormat.setOutputPath(job, new Path(args[1]));
        
        // 对输出结果设置压缩
        FileOutputFormat.setCompressOutput(job, true);
        // 设置压缩类型
        FileOutputFormat.setOutputCompressorClass(job, GzipCodec.class);
        
        return job.waitForCompletion(true) ? 0 : 1;
    }
    
    public static void main(String[] args) throws Exception {        
        String[] date = {"20120917","20120918","20120919","20120920","20120921","20120922","20120923"};
        
        int result = 1;
        for(String dt : date){
            String[] args0 = { 
                    "hdfs://hadoop1:9000/buaa/tv/" + dt + ".txt",
                    "hdfs://hadoop1:9000/buaa/tv/out/"+dt
                };
            result = ToolRunner.run(new Configuration(), new CountUserData(), args0);
        }
        
        System.exit(result);
    }
}