<!--hbase 所需要的架包-->
<dependency>
<groupId>org.apache.hbase</groupId>
<artifactId>hbase-client</artifactId>
<version>1.3.1</version>
</dependency>
<dependency>
<groupId>org.apache.hbase</groupId>
<artifactId>hbase-server</artifactId>
<version>1.3.1</version>
</dependency>
<dependency>
<groupId>org.apache.hbase</groupId>
<artifactId>hbase-common</artifactId>
<version>1.3.1</version>
</dependency>
<dependency>
<groupId>commons-logging</groupId>
<artifactId>commons-logging</artifactId>
<version>1.2</version>
</dependency>
<dependency>
<groupId>log4j</groupId>
<artifactId>log4j</artifactId>
<version>1.2.17</version>
</dependency>
<!-- JSON工具类 -->
<dependency>
<groupId>com.alibaba</groupId>
<artifactId>fastjson</artifactId>
<version>1.2.68</version>
</dependency>
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<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) 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<Map<String,Object>> list=getResultList(accessArray);
for (Map<String,Object> map:list){
//设置一条数据的rowkey值
String rowKey=System.currentTimeMillis()+"";
for (Map.Entry<String,Object> entry:map.entrySet()){
insertRow(initHbase(),tableName,rowKey,"i",entry.getKey(),entry.getValue().toString());
}
}
}
}