共享单车之数据分析 第3关：统计共享单车指定车辆每次使用的空闲平均时间

package com.educoder.bigData.sharedbicycle;

import java.io.IOException;

import java.math.BigDecimal;

import java.math.RoundingMode;

import org.apache.hadoop.conf.Configuration;

import org.apache.hadoop.conf.Configured;

import org.apache.hadoop.hbase.CompareOperator;

import org.apache.hadoop.hbase.client.Put;

import org.apache.hadoop.hbase.client.Result;

import org.apache.hadoop.hbase.client.Scan;

import org.apache.hadoop.hbase.filter.Filter;

import org.apache.hadoop.hbase.filter.SingleColumnValueFilter;

import org.apache.hadoop.hbase.io.ImmutableBytesWritable;

import org.apache.hadoop.hbase.mapreduce.TableMapReduceUtil;

import org.apache.hadoop.hbase.mapreduce.TableMapper;

import org.apache.hadoop.hbase.mapreduce.TableReducer;

import org.apache.hadoop.hbase.util.Bytes;

import org.apache.hadoop.io.BytesWritable;

import org.apache.hadoop.io.Text;

import org.apache.hadoop.mapreduce.Job;

import org.apache.hadoop.util.Tool;

import com.educoder.bigData.util.HbaseUtil;

public class FreeTimeMapReduce extends Configured implements Tool {

    public static final byte[] family = "info".getBytes();

    public static class MyMapper extends TableMapper {

        protected void map(ImmutableBytesWritable rowKey, Result result, Context context)

                throws IOException, InterruptedException {

            long beginTime = Long.parseLong(Bytes.toString(result.getValue(family, "beginTime".getBytes())));

            long endTime = Long.parseLong(Bytes.toString(result.getValue(family, "endTime".getBytes())));

            BytesWritable bytesWritable = new BytesWritable(Bytes.toBytes(beginTime + "_" + endTime));

            context.write(new Text("5996"), bytesWritable);      

        }

    }

    public static class MyTableReducer extends TableReducer {

        @Override

        public void reduce(Text key, Iterable values, Context context)

                throws IOException, InterruptedException {

            long freeTime = 0;

            long beginTime = 0;

            int length = 0;

            for (BytesWritable time : values) {

                byte[] copyBytes = time.copyBytes();

                String timeLong = Bytes.toString(copyBytes);

                String[] split = timeLong.split("_");

                if(beginTime == 0) {

                    beginTime = Long.parseLong(split[0]);

                    continue;

                }

                else {

                    freeTime = freeTime + beginTime - Long.parseLong(split[1]);

                    beginTime = Long.parseLong(split[0]);

                    length ++;

                }

            }

            Put put = new Put(Bytes.toBytes(key.toString()));

            BigDecimal decimal = new BigDecimal(freeTime / length /1000 /60 /60);

            BigDecimal setScale = decimal.setScale(2, RoundingMode.HALF_DOWN);

            put.addColumn(family, "freeTime".getBytes(), Bytes.toBytes(setScale.toString()));

            context.write(null, put);

        }

    }

    public int run(String[] args) throws Exception {

        // 配置Job

        Configuration conf = HbaseUtil.conf;

        // Scanner sc = new Scanner(System.in);

        // String arg1 = sc.next();

        // String arg2 = sc.next();

        String arg1 = "t_shared_bicycle";

        String arg2 = "t_bicycle_freetime";

        try {

            HbaseUtil.createTable(arg2, new String[] { "info" });

        } catch (Exception e) {

            // 创建表失败

            e.printStackTrace();

        }

        Job job = configureJob(conf, new String[] { arg1, arg2 });

        return job.waitForCompletion(true) ? 0 : 1;

    }

    private Job configureJob(Configuration conf, String[] args) throws IOException {

        String tablename = args[0];

        String targetTable = args[1];

        Job job = new Job(conf, tablename);

        Scan scan = new Scan();

        scan.setCaching(300);

        scan.setCacheBlocks(false);// 在mapreduce程序中千万不要设置允许缓存

         //设置过滤条件

        Filter filter = new SingleColumnValueFilter(Bytes.toBytes("info"), Bytes.toBytes("bicycleId"), CompareOperator.EQUAL, Bytes.toBytes("5996"));

        scan.setFilter(filter); 

        // 初始化Mapreduce程序

        TableMapReduceUtil.initTableMapperJob(tablename, scan, MyMapper.class, Text.class, BytesWritable.class, job);

        // 初始化Reduce

        TableMapReduceUtil.initTableReducerJob(targetTable, // output table

                MyTableReducer.class, // reducer class

                job);

        job.setNumReduceTasks(1);

        return job;

    }

}