E-MapReduce Spark + Hbase是什么?

Spark + Hbase



Spark 接入 Hbase


下面这个例子演示了 Spark 如何向 Hbase 写数据。需要指出的是，计算集群需要和 Hbase 集群处于一个安全组内，否则网络无法打通。在E-Mapreduce 创建集群时，请注意选择 Hbase 集群所处的安全组。

1.     object ConnectionUtil extends Serializable {
2.       private val conf = HBaseConfiguration.create()
3.       conf.set(HConstants.ZOOKEEPER_QUORUM,"ecs1,ecs1,ecs3")
4.       conf.set(HConstants.ZOOKEEPER_ZNODE_PARENT, "/hbase")
5.       private val connection = ConnectionFactory.createConnection(conf)
7.       def getDefaultConn: Connection = connection
8.     }
10.     //创建数据流 unionStreams
11.     unionStreams.foreachRDD(rdd => {
12.       rdd.map(bytes => new String(bytes))
13.         .flatMap(line => line.split(" "))
14.         .map(word => (word, 1))
15.         .reduceByKey(_ + _)
16.         .mapPartitions {words => {
17.           val conn = ConnectionUtil.getDefaultConn
18.           val tableName = TableName.valueOf(tname)
19.           val t = conn.getTable(tableName)
20.           try {
21.             words.sliding(100, 100).foreach(slice => {
22.               val puts = slice.map(word => {
23.                 println(s"word: $word")
24.                 val put = new Put(Bytes.toBytes(word._1 + System.currentTimeMillis()))
25.                 put.addColumn(COLUMN_FAMILY_BYTES, COLUMN_QUALIFIER_BYTES,
26.                   System.currentTimeMillis(), Bytes.toBytes(word._2))
27.                 put
28.               }).toList
29.               t.put(puts)
30.             })
31.           } finally {
32.             t.close()
33.           }
35.           Iterator.empty
36.         }}.count()
37.     })
39.     ssc.start()
40.     ssc.awaitTermination()

• Spark接入 Hbase