json数据写入hbase

      
        
            org.apache.hbase
            hbase-client
            1.3.1
        
        
            org.apache.hbase
            hbase-server
            1.3.1
        
        
            org.apache.hbase
            hbase-common
            1.3.1
        

        
            commons-logging
            commons-logging
            1.2
        

        
            log4j
            log4j
            1.2.17
        


        

        
            com.alibaba
            fastjson
            1.2.68
        

package main.scala.com.web.zhangyong168.cn.spark.java;
import com.alibaba.fastjson.JSONObject;
import org.apache.hadoop.hbase.HBaseConfiguration;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.*;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.util.Bytes;
import java.io.IOException;
import java.util.ArrayList;
import java.util.LinkedHashMap;
import java.util.List;
import java.util.Map;

/**
 * @author zhangyong
 * @version 1.0.0
 * @description : json数据写入hbase
 * @date 2020/05/28  10:55
 */
public class WriteHbase {
    /**
     * 获取hbase hadoop 的相关配置文件
     *
     * @return
     * @throws IOException
     */
    public static Connection initHbase() throws IOException {
        Configuration configuration = HBaseConfiguration.create();
        configuration.addResource("hadoop/mapred-site.xml");
        configuration.addResource("hadoop/hive-site.xml");
        configuration.addResource("hadoop/hive-site.xml");
        configuration.addResource("hadoop/ssl-client.xml");
        configuration.addResource("hadoop/yarn-site.xml");
        configuration.addResource("hadoop/hbase-site.xml");
        configuration.addResource("hadoop/core-site.xml");
        Connection connection = ConnectionFactory.createConnection(configuration);
        return connection;
    }


    /**
     * @param connection hbase连接
     * @param tableName  表名（hbase表空间加表名）
     * @param rowKey     rowKey值主值
     * @param colFamliy  列族 默认为i
     * @param col        列名
     * @param val        列值
     */
    public static void insertRow(Connection connection, String tableName, String rowKey, String colFamliy, String col, String val) throws IOException {
        Table table = connection.getTable(TableName.valueOf(tableName));
        Put put = new Put(Bytes.toBytes(rowKey));
        put.addColumn(Bytes.toBytes(colFamliy), Bytes.toBytes(col), Bytes.toBytes(val));
        table.put(put);
        table.close();

    }


    /***
     * 构造数据结果集
     * @param accessArray
     * @return
     */
    public static List> getResultList(AccessArray accessArray) {
        List> list = new ArrayList<>();
        int columnNameLengths = accessArray.getColumnNames().length;
        for (Object[] tempValue : accessArray.getRecordArrayValue()) {
            Map parameters = new LinkedHashMap<>();
            if (columnNameLengths == tempValue.length) {
                for (int j = 0; j                     parameters.put(accessArray.getColumnName(j), tempValue[j].toString());
                }
            }
            list.add(parameters);
        }
        return list;
    }


    public static void main(String[] args) throws IOException {
      String json="{\"columnNames\":[\"id\",\"name\",\"age\",\"birthday\"]," +
          "\"columnsTypes\":[0,0,0,0],\"valuesSize\":2,"  +
          "\"columnValues\":[[\"1\",\"zhagnsan\",\"19\",\"2020-02-20\"],"  +
          "[\"2\",\"lisi\",\"24\",\"2020-02-15\"]]}";

         String tableName="AUEIC.DANIEL";
          AccessArray  accessArray= JSONObject.parseObject(json,AccessArray.class);
        List> list=getResultList(accessArray);
        for (Map map:list){
            //设置一条数据的rowkey值
            String rowKey=System.currentTimeMillis()+"";
           for (Map.Entry entry:map.entrySet()){
            insertRow(initHbase(),tableName,rowKey,"i",entry.getKey(),entry.getValue().toString());
           }
        }
    }
}