@[toc]
參考文檔 : https://hbase.apache.org/book.html#_sparksql_dataframes
簡介
hbase-spark integration使用了Spark-1.2.0中引入的DataSource API (SPARK-3247), 它在簡單的HBase KV存儲和復雜的關系SQL查詢之間架起橋梁，使用戶能夠使用Spark在HBase上執(zhí)行復雜的數(shù)據(jù)分析工作。HBase數(shù)據(jù)幀是一個標準的Spark數(shù)據(jù)幀，能夠與Hive、ORC、Parquet、JSON等任何其他數(shù)據(jù)源交互。HBase Spark集成應用了諸如分區(qū)修剪、列修剪、謂詞下推和數(shù)據(jù)位置等關鍵技術。
要使用hbase-spark integration connector，用戶需要為HBase和Spark表之間的模式映射定義Catalog，準備數(shù)據(jù)并填充HBase表，然后加載HBase數(shù)據(jù)幀。之后，用戶可以使用SQL查詢來集成查詢和訪問HBase表中的記錄。
打包生成hbase-spark庫
使用hbase-spark integration需要hbase-spark庫
找了半天沒有找到最新的那個包, 所以自己去github上面下載代碼打包, 然后安裝到本地倉庫
git clone https://github.com/apache/hbase-connectors.git
cd hbase-connectors/spark/hbase-spark
mvn -Dspark.version=2.4.3 -Dscala.version=2.11.7 -Dscala.binary.version=2.11 clean install
然后在項目pom.xml中添加依賴
<dependency>
<groupId>org.apache.hbase.connectors.spark</groupId>
<artifactId>hbase-spark</artifactId>
<version>1.0.1</version>
</dependency>
<dependency>
<groupId>org.apache.hbase</groupId>
<artifactId>hbase-client</artifactId>
<version>2.1.4</version>
</dependency>
解決訪問Hbase問題
執(zhí)行代碼時出現(xiàn)錯誤:
Exception in thread "main" java.lang.reflect.InvocationTargetException
at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
at sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62)
at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
at java.lang.reflect.Method.invoke(Method.java:498)
at org.apache.spark.deploy.worker.DriverWrapper$.main(DriverWrapper.scala:65)
at org.apache.spark.deploy.worker.DriverWrapper.main(DriverWrapper.scala)
Caused by: java.lang.NoClassDefFoundError: org/apache/hadoop/hbase/fs/HFileSystem
at cn.com.sjfx.sparkappdemo.Application.main(Application.java:27)
... 6 more
Caused by: java.lang.ClassNotFoundException: org.apache.hadoop.hbase.fs.HFileSystem
at java.net.URLClassLoader.findClass(URLClassLoader.java:381)
at java.lang.ClassLoader.loadClass(ClassLoader.java:424)
at java.lang.ClassLoader.loadClass(ClassLoader.java:357)
... 7 more
這是因為spark無法訪問hbase中的庫造成的, 需要在制作鏡像的時候把hbase的庫加入到spark中,
修改Dockerfile, 增加如下內容:
COPY /hbase-lib/* /spark/jars/
讀寫Hbase
public class Application {
public static void main(String[] args) {
SparkConf sparkConf = new SparkConf().setAppName("demo");
JavaSparkContext jsc = new JavaSparkContext(sparkConf);
SparkSession sparkSession = SparkSession.builder()
.sparkContext(jsc.sc())
.getOrCreate();
//設置要訪問的hbase的zookeeper
Configuration configuration = HBaseConfiguration.create();
configuration.set("hbase.zookeeper.quorum", "192.168.1.22:15301,192.168.1.22:15302,192.168.1.22:15303");
//一定要創(chuàng)建這個hbaseContext, 因為后面寫入時會用到它
HBaseContext hBaseContext=new HBaseContext(jsc.sc(),configuration,null);
//創(chuàng)建一個測試用的RDD
List<Integer> data = new ArrayList<>();
for (int i = 0; i < 256; i++) {
data.add(i);
}
JavaRDD<Integer> rdd = jsc.parallelize(data);
JavaRDD<HBaseRecord> rdd1 = rdd.map(i -> new HBaseRecord(i, "extra"));
rdd1.collect().forEach(System.out::println);
//根據(jù)RDD創(chuàng)建數(shù)據(jù)幀
Dataset<Row> df = sparkSession.createDataFrame(rdd1, HBaseRecord.class);
//定義映射的catalog
String catalog = "{" +
" \"table\":{\"namespace\":\"default\", \"name\":\"table1\"}," +
" \"rowkey\":\"key\"," +
" \"columns\":{" +
" \"col0\":{\"cf\":\"rowkey\", \"col\":\"key\", \"type\":\"string\"}," +
" \"col1\":{\"cf\":\"cf1\", \"col\":\"col1\", \"type\":\"boolean\"}," +
" \"col2\":{\"cf\":\"cf2\", \"col\":\"col2\", \"type\":\"double\"}," +
" \"col3\":{\"cf\":\"cf3\", \"col\":\"col3\", \"type\":\"float\"}," +
" \"col4\":{\"cf\":\"cf4\", \"col\":\"col4\", \"type\":\"int\"}," +
" \"col5\":{\"cf\":\"cf5\", \"col\":\"col5\", \"type\":\"bigint\"}," +
" \"col6\":{\"cf\":\"cf6\", \"col\":\"col6\", \"type\":\"smallint\"}," +
" \"col7\":{\"cf\":\"cf7\", \"col\":\"col7\", \"type\":\"string\"}," +
" \"col8\":{\"cf\":\"cf8\", \"col\":\"col8\", \"type\":\"tinyint\"}" +
" }" +
" }";
//寫入數(shù)據(jù)
df.write()
.format("org.apache.hadoop.hbase.spark")
.option(HBaseTableCatalog.tableCatalog(), catalog)
.option(HBaseTableCatalog.newTable(), "5") //寫入到5個分區(qū)
.mode(SaveMode.Overwrite) // 覆蓋模式
.save();
//讀取數(shù)據(jù)
Dataset<Row> df2 = sparkSession.read()
.format("org.apache.hadoop.hbase.spark")
.option(HBaseTableCatalog.tableCatalog(), catalog)
.load();
System.out.println("read result: ");
df2.show();
}
//類需要可序列化
public static class HBaseRecord implements Serializable {
private static final long serialVersionUID = 4331526295356820188L;
//屬性一定要getter/setter, 即使是public
public String col0;
public Boolean col1;
public Double col2;
public Float col3;
public Integer col4;
public Long col5;
public Short col6;
public String col7;
public Byte col8;
public String getCol0() {
return col0;
}
public void setCol0(String col0) {
this.col0 = col0;
}
public Boolean getCol1() {
return col1;
}
public void setCol1(Boolean col1) {
this.col1 = col1;
}
public Double getCol2() {
return col2;
}
public void setCol2(Double col2) {
this.col2 = col2;
}
public Float getCol3() {
return col3;
}
public void setCol3(Float col3) {
this.col3 = col3;
}
public Integer getCol4() {
return col4;
}
public void setCol4(Integer col4) {
this.col4 = col4;
}
public Long getCol5() {
return col5;
}
public void setCol5(Long col5) {
this.col5 = col5;
}
public Short getCol6() {
return col6;
}
public void setCol6(Short col6) {
this.col6 = col6;
}
public String getCol7() {
return col7;
}
public void setCol7(String col7) {
this.col7 = col7;
}
public Byte getCol8() {
return col8;
}
public void setCol8(Byte col8) {
this.col8 = col8;
}
public HBaseRecord(Integer i, String s) {
col0 = String.format("row%03d", i);
col1 = i % 2 == 0;
col2 = Double.valueOf(i);
col3 = Float.valueOf(i);
col4 = i;
col5 = Long.valueOf(i);
col6 = i.shortValue();
col7 = "String:" + s;
col8 = i.byteValue();
}
@Override
public String toString() {
return "HBaseRecord{" +
"col0='" + col0 + '\'' +
", col1=" + col1 +
", col2=" + col2 +
", col3=" + col3 +
", col4=" + col4 +
", col5=" + col5 +
", col6=" + col6 +
", col7='" + col7 + '\'' +
", col8=" + col8 +
'}';
}
}
}
