package main.scala.com.web.zhangyong168.cn.spark.java;
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 {

    private Admin admin;

    /**
     * 获取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<Map<String, Object>> getResultList(AccessArray accessArray) {
        List<Map<String, Object>> list = new ArrayList<>();
        int columnNameLengths = accessArray.getColumnNames().length;
        for (Object[] tempValue : accessArray.getRecordArrayValue()) {
            Map<String, Object> parameters = new LinkedHashMap<>();
            if (columnNameLengths == tempValue.length) {
                for (int j = 0; j <= columnNameLengths; j++) {
                    parameters.put(accessArray.getColumnName(j), tempValue[j].toString());
                }
            }
            list.add(parameters);
        }
        return list;
    }


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