public class WriteHBase {

    public static void writeHbase(String content){

　　　　 // HDFS 数据是一行一条记录

        String[] lines = content.split("\n");

        int userSize = 0;

        List<Put> puts = new ArrayList<Put>();

        Put put;

        for(String line : lines){
　　　　　　  //只有两列,以#号分割，一列rowkey，一列value，一个value是很多列数据拼接起来的。

            if(line.contains("#")){

                String[] arr = line.split("#");

//              添加一行，

                put = new Put(Bytes.toBytes(arr[0]));
　　　　　　　　　// 给行添加列 cf column value

                put.add(Bytes.toBytes(Constant.CF), Bytes.toBytes(Constant.COLUMN), Bytes.toBytes(arr[1]));

                puts.add(put);

            }else{

                continue;

            }

            lines[userSize] = null;

            ++userSize;

            // write when list have 1000 没1000 条提交一次，已经改的 5000

            if (userSize % Constant.BATCH ==0){

                writeDate(userSize, puts);

            }

        }

        writeDate(userSize, puts);

        HDFSReadLog.writeLog("analysis " +userSize +" users");

    }

    private static void writeDate(int userSize, List<Put> puts) {

        try {

            table.put(puts);

            HDFSReadLog.writeLog("write "+userSize + " item.");

        } catch (IOException e) {

            e.printStackTrace();

            HDFSReadLog.writeLog("write "+userSize + " error.");

            HDFSReadLog.writeLog(e.getMessage());

        }

    }

    static HTable table = null;

//    static HTablePool pool = null;

    static{

        try {
　　　　　　　　// 创建HTable对象，对应hbase 的table

            table = new HTable(HBaseConf.getConf(),Constant.HBASE_TABLE);
　　　　　　　　// 如果表不存在就创建一个

            fitTable(Constant.HBASE_TABLE);

        } catch (IOException e) {

            e.printStackTrace();

            HDFSReadLog.writeLog("create table error.");

            HDFSReadLog.writeLog(e.getMessage());

        }

    }

    /**

     * if table is not exists, create it

     * @param tab

     * @throws IOException

     */

    private static void fitTable(String tab) throws IOException {

        HBaseAdmin admin = new HBaseAdmin(HBaseConf.getConf());

        if (admin.tableExists(tab)) {

            HDFSReadLog.writeLog(tab + " exists");

        } else {
　　　　　　　　

            HTableDescriptor tableDesc = new HTableDescriptor(tab);
　　　　　　　　// 建表的使用要指定 column family

            tableDesc.addFamily(new HColumnDescriptor("cf"));

            admin.createTable(tableDesc);

            HDFSReadLog.writeLog(tab + " create success");

        }

    }

}

public class HBaseConf {

    public static Configuration conf = null;

    public static Configuration getConf(){

        if (conf == null){

            conf = new Configuration();

            String path  = Constant.getSysEnv("HBASE_HOME") +"/conf/";

            HDFSReadLog.writeLog("Get HBase home : " + path);

            // hbase conf

            conf.setClassLoader(HBaseConf.class.getClassLoader());

            conf.addResource(path + "hbase-default.xml");

            conf.addResource(path + "hbase-site.xml");

            conf = HBaseConfiguration.create(conf);

            HDFSReadLog.writeLog("hbase.zookeeper.quorum : " + conf.get("hbase.zookeeper.quorum"));

        }

　　　　// 如果配置文件读不到，set这两个参数，也可以读

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

        conf.set("hbase.zookeeper.property.clientPort", "port");*/

        return conf;

    }

}