mapreduce实现从hbase中统计数据，结果存入mysql中

最近开始学习使用mapreduce统计hbase中的数据，并将结果集存入mysql中，供前台查询使用。

使用hadoop版本为2.5.1，hbase版本为0.98.6.1

mapreduce程序分为三个部分：job、map函数、reduce函数

job类：

public class DayFaultStatisticsJob {
    private static final Logger logger = LoggerFactory.getLogger(DayFaultStatisticsJob.class);

    public void runJob(String start){
        try {
            logger.info("开始运行mapreduce，统计故障信息");
            Configuration config = HBaseConfiguration.create();
            // 传递统计条件给map
            config.set("search.time.start",start);

            DBConnector dbConnector = DBConnectorUtil.getDBConnector();
            DBConfiguration.configureDB(config, "com.mysql.jdbc.Driver", "jdbc:mysql://" + dbConnector.getHost() + ":" + dbConnector.getPort() + "/xcloud", dbConnector.getUser(), dbConnector.getPwd());

            Job job = Job.getInstance(config);
            job.setJarByClass(DayFaultStatisticsJob.class); // class that contains mapper and reducer
            Scan scan = new Scan();
            scan.setCaching(500);        // 1 is the default in Scan, which will be bad for MapReduce jobs
            scan.setCacheBlocks(false);  // don't set to true for MR jobs
            scan.addFamily(GlobalConstants.DEVICE_FAULT_FAMILY_NAME);

            // set other scan attrs
            // TODO 增加遍历条件
            /*scan.setStartRow();
              scan.setStopRow();*/
            //scan.setFilter();

            TableMapReduceUtil.initTableMapperJob(
                    GlobalConstants.TABLE_NAME_DEVICE_FAULT,          // input table
                    scan,                // Scan instance to control CF and attribute selection
                    DayFaultStatisticsMapper.class,     // mapper class
                    Text.class,         // mapper output key
                    IntWritable.class,  // mapper output value
                    job);
            job.setReducerClass(DayFaultStatisticsReducer.class);    // reducer class
            job.setNumReduceTasks(2);    // at least one, adjust as required
            //FileOutputFormat.setOutputPath(job, new Path("/usr/local/mapreduce"));  // adjust directories as required
            DBOutputFormat.setOutput(job, GlobalConstants.MYSQL_DEVICE_FAULT_DAY, GlobalConstants.MYSQL_DEVICE_FAULT_DAY_FIELDS);
            boolean b = job.waitForCompletion(true);
            if (b) {
                logger.info("mapreduce任务正常结束");
                System.exit(0);
            } else {
                logger.info("mapreduce任务异常结束");
                System.exit(1);
            }
        } catch (IOException e) {
            logger.error(e.getMessage(),e);
        } catch (InterruptedException e) {
            logger.error(e.getMessage(),e);
        } catch (ClassNotFoundException e) {
            logger.error(e.getMessage(),e);
        }
    }

}

mapper类：

public class DayFaultStatisticsMapper extends TableMapper<Text, IntWritable> {
    private static final Logger logger = LoggerFactory.getLogger(DayFaultStatisticsMapper.class);

    private Text text = new Text();
    private IntWritable ONE = new IntWritable(1);
    HashMap<String,String> conditionMap = new HashMap<String,String>();

    // 接收过滤条件
    protected void setup(Context context) throws IOException,
            InterruptedException {
        conditionMap.put("start",context.getConfiguration().get("search.time.start").trim());
    }

    public void map(ImmutableBytesWritable key, Result value, Context context) throws IOException, InterruptedException {
        String timestamp = null;
        String company_id = null;
        String product_id = null;
        String model_id = null;
        String fault_id = null;
        String province_id = null;
        String city_id = null;
        String district_id = null;
        for(java.util.Map.Entry<byte[], byte[]> val : value.getFamilyMap(Bytes.toBytes("info")).entrySet()){
            String ikey = new String(val.getKey());
            String ivalue = new String(val.getValue());
            // 按条件进行map
            if (StringUtil.isEmpty(ikey) || StringUtil.isEmpty(ivalue)){
                 return;
            } else if (ikey.equals("company_id")){
                company_id = ivalue;
                continue;
            } else if (ikey.equals("product_id")){
                product_id = ivalue;
                continue;
            } else if (ikey.equals("model_id")){
                model_id = ivalue;
                continue;
            } else if (ikey.equals("fault_id")){
                fault_id = ivalue;
                continue;
            } else if (ikey.equals("province_id")){
                province_id = ivalue;
                continue;
            } else if (ikey.equals("city_id")){
                city_id = ivalue;
                continue;
            } else if (ikey.equals("district_id")){
                district_id = ivalue;
                continue;
            } else if (ikey.equals("timestamp")){
                String time = ivalue.substring(0,8);// 判断是否是当日
                if (time.equals(conditionMap.get("start"))){
                    timestamp = time; // 统计日故障发生次数
                }
                continue;
            }
        }
        if (company_id == null || product_id ==  null || model_id == null || fault_id == null || timestamp == null || fault_id.equals("-1")) {
            return;
        }
        // 故障码为 -1的不统计
        String val = company_id + "-" +product_id + "-" +model_id + "-"+fault_id + "-"+timestamp + "-"+province_id + "-"+ city_id+ "-"+ district_id; // 分组key
        text.set(val);     // we can only emit Writables...
        context.write(text, ONE);
    }
}

reducer类：

public class DayFaultStatisticsReducer extends Reducer<Text, IntWritable, FaultDay, Text> {
    private static final Logger logger = LoggerFactory.getLogger(DayFaultStatisticsReducer.class);

    public void reduce(Text key, Iterable<IntWritable> values, Context context) throws IOException, InterruptedException {
        int value = 0;
        for (IntWritable val : values) {
            value += val.get();
        }
        //company_id + "-" +product_id + "-" +model_id + "-"+fault_id + "-"+timestamp + "-"+province_id + "-"+ city_id+ "-"+ district_id
        String fields[] = key.toString().split("-");
        if (fields.length != 8){
            return;
        }
        // 处理day day格式：2015-02-01
        String day = fields[4].substring(0,4) + "-"  + fields[4].substring(4,6) + "-" + fields[4].substring(6,8);
        // DateUtil.format(new Date(), new SimpleDateFormat("yyyy-MM-dd'T'HH:mm:ss"));
        FaultDay faultDay = new FaultDay();
        faultDay.setId(UUIDGen.generate());
        faultDay.setDay(day);
        faultDay.setCompany_id(fields[0]);
        faultDay.setProduct_id(fields[1]);
        faultDay.setModel_id(fields[2]);
        faultDay.setFault_id(fields[3]);
        if (!fields[5].equals("null")){
            faultDay.setProvince_id(fields[5]);
        }
        if (!fields[6].equals("null")){
            faultDay.setCity_id(fields[6]);
        }
        if (!fields[7].equals("null")){
            faultDay.setDistrict_id(fields[7]);
        }
        faultDay.setNum(value);
        context.write(faultDay, null);
    }

}

FaultDay中实现DBWritable接口

public class FaultDay implements Writable, DBWritable {
    private String id;

    private String day; // 2015-02-01
    private String company_id;
    private String product_id;
    private String model_id;
    private String fault_id;
    private String province_id;
    private String city_id;
    private String district_id;
    private int num;

    @Override
    public void write(PreparedStatement statement) throws SQLException {
        int index = 1;
        statement.setString(index++, this.getId());
        statement.setString(index++, this.getDay());
        statement.setString(index++, this.getCompany_id());
        statement.setString(index++, this.getProduct_id());
        statement.setString(index++, this.getModel_id());
        statement.setString(index++, this.getFault_id());
        statement.setString(index++, this.getProvince_id());
        statement.setString(index++, this.getCity_id());
        statement.setString(index++, this.getDistrict_id());
        statement.setInt(index++, this.getNum());
    }

    @Override
    public void readFields(ResultSet resultSet) throws SQLException {
        this.id = resultSet.getString(1);
        this.day = resultSet.getString(2);
        this.company_id = resultSet.getString(3);
        this.product_id = resultSet.getString(4);
        this.model_id = resultSet.getString(5);
        this.fault_id = resultSet.getString(6);
        this.province_id = resultSet.getString(7);
        this.city_id = resultSet.getString(8);
        this.district_id = resultSet.getString(9);
        this.num = resultSet.getInt(10);
    }

    @Override
    public void write(DataOutput out) throws IOException {
        //To change body of implemented methods use File | Settings | File Templates.
    }

    @Override
    public void readFields(DataInput in) throws IOException {
        //To change body of implemented methods use File | Settings | File Templates.
    }


    public String getCity_id() {
        return city_id;
    }

    public void setCity_id(String city_id) {
        this.city_id = city_id;
    }

    public String getCompany_id() {
        return company_id;
    }

    public void setCompany_id(String company_id) {
        this.company_id = company_id;
    }

    public String getDay() {
        return day;
    }

    public void setDay(String day) {
        this.day = day;
    }

    public String getDistrict_id() {
        return district_id;
    }

    public void setDistrict_id(String district_id) {
        this.district_id = district_id;
    }

    public String getFault_id() {
        return fault_id;
    }

    public void setFault_id(String fault_id) {
        this.fault_id = fault_id;
    }

    public String getId() {
        return id;
    }

    public void setId(String id) {
        this.id = id;
    }

    public String getModel_id() {
        return model_id;
    }

    public void setModel_id(String model_id) {
        this.model_id = model_id;
    }

    public int getNum() {
        return num;
    }

    public void setNum(int num) {
        this.num = num;
    }

    public String getProduct_id() {
        return product_id;
    }

    public void setProduct_id(String product_id) {
        this.product_id = product_id;
    }

    public String getProvince_id() {
        return province_id;
    }

    public void setProvince_id(String province_id) {
        this.province_id = province_id;
    }
}

主类：

public class FaultStatistics {
    private static final Logger logger = LoggerFactory.getLogger(FaultStatistics.class);

    public static void main(String[] args) {
        // 传递统计条件给map
        String start;
        if ( args.length != 0 && StringUtil.notEmpty(args[0])) {
            start = args[0].substring(0,8);
        } else {
            // 默认为当前时间的前一天
            start = DateUtil.format(DateUtil.getPreDay(new Date()), new SimpleDateFormat("yyyyMMdd"));
        }
        
        DayFaultStatisticsJob dayFaultStatisticsJob = new DayFaultStatisticsJob();
        dayFaultStatisticsJob.runJob(start);
    }

}

打成jar包，hadoop中运行，可在mysql中查询到运行结果（mysql中要存在对应的表）