1. 程式人生 > >spark(2.1.0) 操作hbase(1.0.2)

spark(2.1.0) 操作hbase(1.0.2)

hadoop mon per bsp trac 事先 com maker scala

1、spark中引入外部jar包

  1)創建/usr/software/spark_jars目錄,放入spark操作hbase的jar包:hbase-annotations-1.0.2.jar,hbase-client-1.0.2.jar,hbase-common-     1.0.2.jar,hbase-server-1.0.2.jar

  2)修改spark-default.conf文件,加入以下兩行: 

    spark.executor.extraClassPath=/usr/software/spark_jars/*
    spark.driver.extraClassPath=/usr/software/spark_jars/*

2、進入hbase事先創建好表

    create ‘test‘,‘f1‘

2、進行spark-shell或用scala進行操作hbase。

3、代碼部分:

import org.apache.hadoop.hbase.{HBaseConfiguration, TableName}
import org.apache.hadoop.hbase.client.{ConnectionFactory, Put}
import org.apache.hadoop.hbase.io.ImmutableBytesWritable
import org.apache.hadoop.hbase.mapred.TableOutputFormat
import org.apache.hadoop.hbase.util.Bytes
import org.apache.hadoop.mapred.JobConf
import org.apache.spark.{SparkConf, SparkContext}
import org.apache.spark.streaming.{Seconds, StreamingContext}

val conf = HBaseConfiguration.create()
var jobConf = new JobConf(conf)
jobConf.set("hbase.zookeeper.quorum", "localhost")
jobConf.set("zookeeper.znode.parent", "/hbase")
jobConf.set(TableOutputFormat.OUTPUT_TABLE, "test")
jobConf.setOutputFormat(classOf[TableOutputFormat])
val rdd = sc.makeRDD(Array(1)).flatMap(_ => 0 to 100000)
rdd.map(x => {
var put = new Put(Bytes.toBytes(x.toString))
put.addColumn(Bytes.toBytes("f1"), Bytes.toBytes("c1"), Bytes.toBytes(x.toString))
(new ImmutableBytesWritable, put)
}).saveAsHadoopDataset(jobConf)

spark(2.1.0) 操作hbase(1.0.2)