/**
 * Created by Administrator on 2017/8/18.
 */
public class IteblogBulkLoadDriver {
    public static class IteblogBulkLoadMapper extends Mapper<LongWritable, Text, StringWriter, Put> {
        protected void map(LongWritable key, Text value, Context context) throws InterruptedException, IOException {
            if(value==null) {
                return;
            }

            String line = value.toString();

            String[] items = line.split("\\^");
            if(items.length<3){
                items = line.split("\\^");
            }
            if(items.length<3){
                System.out.println("================less 3");
                return;
            }
            System.out.println(line);
            String rowKey = items[0]+items[1];
            Put put = new Put(Bytes.toBytes(items[0]));   //ROWKEY
            put.addColumn("cf".getBytes(), "url".getBytes(), items[1].getBytes());
            put.addColumn("cf".getBytes(), "name".getBytes(), items[2].getBytes());
            context.write(new StringWriter().append(rowKey), put);
        }
    }



    public static class HBaseHFileReducer extends
            Reducer<StringWriter, Put, ImmutableBytesWritable, Put> {
        protected void reduce(StringWriter key, Iterable<Put> values,
                              Context context) throws IOException, InterruptedException {
            String value = "";
            ImmutableBytesWritable k  = new ImmutableBytesWritable(key.toString().getBytes());

            Put val = values.iterator().next();
            context.write(k, val);
        }


    }

    public static void main(String[] args) throws IOException, ClassNotFoundException, InterruptedException {
//              String SRC_PATH= "hdfs:/slave1:8020/maats5/pay/logdate=20170906";
//              String DESC_PATH= "hdfs:/slave1:8020/maats5_test/pay/logdate=20170906";
            String SRC_PATH= args[0];
            String DESC_PATH=args[1];
            Configuration conf = HBaseConnectionFactory.config;
            conf.set("fs.hdfs.impl", "org.apache.hadoop.hdfs.DistributedFileSystem");
            Job job=Job.getInstance(conf);
            job.setJarByClass(IteblogBulkLoadDriver.class);
            job.setMapperClass(IteblogBulkLoadMapper.class);
            job.setMapOutputKeyClass(StringWriter.class);
            job.setMapOutputValueClass(Put.class);
            job.setReducerClass(HBaseHFileReducer.class);
            job.setOutputFormatClass(HFileOutputFormat2.class);
            HTable table = new HTable(conf,"maatstest");
            HFileOutputFormat2.configureIncrementalLoad(job,table,table.getRegionLocator());
            FileInputFormat.addInputPath(job,new Path(SRC_PATH));
            FileOutputFormat.setOutputPath(job,new Path(DESC_PATH));

            System.exit(job.waitForCompletion(true)?0:1);
        }
    }