package bulkloadExample;

import org.apache.hadoop.conf.Configuration;

import org.apache.hadoop.fs.FileSystem;

import org.apache.hadoop.fs.Path;

import org.apache.hadoop.hbase.HBaseConfiguration;

import org.apache.hadoop.hbase.TableName;

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

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

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

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

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

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

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

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

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

import org.apache.hadoop.io.LongWritable;

import org.apache.hadoop.io.Text;

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 java.io.IOException;

import java.net.URI;

public class BulkloadData {

    static final String JOBNAME = "BulkLoad";

    static final String TABLENAME = "wqbin:duowan_user_bulkLoad";

    static final String PATH_IN = "/datain/duowan_user.txt";    //输入路径

    static final String PATH_OUT = "/dataout";    //输入路径HFILE

    static final String SEPARATOR = "\t";

    static final byte[] ColumnFamily =  "f".getBytes();  // 列簇

//    static final byte[] row_id =  "id".getBytes();    // 列名

    static final byte[] name =  "name".getBytes();

    static final byte[] pass =  "pass".getBytes();

    static final byte[] mail =  "mail".getBytes();

    static final byte[] nickname =  "nickname".getBytes();

    public static class Map extends Mapper<LongWritable, Text, ImmutableBytesWritable, Put> {

        protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {

            String[] strArr = value.toString().split(SEPARATOR);

            String row = strArr[0];

            Put put = new Put(Bytes.toBytes(row.toString()));     // rowkey

            put.add(ColumnFamily, name, Bytes.toBytes(strArr[1]));

            put.add(ColumnFamily, pass, Bytes.toBytes(strArr[2]));

            put.add(ColumnFamily, mail, Bytes.toBytes(strArr[3]));

            put.add(ColumnFamily, nickname, Bytes.toBytes(strArr[3]));

            context.write(new ImmutableBytesWritable(value.getBytes()), put);

        }

    }

    public static void main(String[] args) throws Exception {

        Configuration conf = HBaseConfiguration.create();

        Connection conn = ConnectionFactory.createConnection(conf);

//        conf.set("hbase.zookeeper.quorum", "xx,xx,xx");

        conf.set("fs.defaultFS","file:///");

        System.setProperty("HADOOP_USER_NAME", "centos");

        Job job =  Job.getInstance(conf, JOBNAME);

        job.setJarByClass(BulkloadData.class);

        job.setMapOutputKeyClass(ImmutableBytesWritable.class);

        job.setMapOutputValueClass(Put.class);

        TableMapReduceUtil.addDependencyJars(job);

        job.setMapperClass(Map.class);

        //这个 SorterReducer(KeyValueSortReducer或PutSortReducer) 可以不指定，

        //因为源码中已经做了判断

        job.setReducerClass(PutSortReducer.class);

        job.setOutputFormatClass(HFileOutputFormat2.class);

        FileSystem fs = FileSystem.get(URI.create("/"),conf);

        Path outPath = new Path(PATH_OUT);

//        if (fs.exists(outPath))fs.delete(outPath, true);

        FileOutputFormat.setOutputPath(job, outPath);

        FileInputFormat.setInputPaths(job, new Path(PATH_IN));

//        HTable table = new HTable(conf, TABLENAME);

        HFileOutputFormat2.configureIncrementalLoad(job, conn.getTable(TableName.valueOf(TABLENAME)),

                conn.getRegionLocator(TableName.valueOf(TABLENAME)));

        System.out.println("load完毕");

        if(job.waitForCompletion(true)){

            LoadIncrementalHFiles loader = new LoadIncrementalHFiles(conf);

//            loader.doBulkLoad( table);

        }

        System.exit(0);

    }

}