pyspark中操作HBase数据以及sbt简单用法

2020-12-19 09:17:03 编辑 1 阅读量:2215

pyspark中操作HBase数据，包括写和读。

一、写数据到HBase

```
# save to hbase    
      host = '192.168.10.23'    
      hbase_table = 'test'
      conf = {"hbase.zookeeper.quorum": host,
              "hbase.mapred.outputtable": hbase_table,
            "mapreduce.outputformat.class": "org.apache.hadoop.hbase.mapreduce.TableOutputFormat",
            "mapreduce.job.output.key.class": "org.apache.hadoop.hbase.io.ImmutableBytesWritable",
              "mapreduce.job.output.value.class": "org.apache.hadoop.io.Writable"
              }          
    keyConv = "org.apache.spark.examples.pythonconverters.StringToImmutableBytesWritableConverter"
    valueConv = "org.apache.spark.examples.pythonconverters.StringListToPutConverter"
       dl= [['sparkrow1', 'cf', 'a', 'value1']]
       sc.parallelize(dl).map(lambda x: (x[0], x))\
         .saveAsNewAPIHadoopDataset(conf, keyConv, valueConv)
```

其中，比较关键的是keyConv, valueConv的定义，这里使用了example里面的converter，这个可以自己写，然后打包成jar。然后，提交任务时需要将jar一并提交，

二、从HBase读数据

流程与写数据类似

```python
# 数据格式：
# scan 'test'
'''
\xE5\xAB\x98\xE7\xA5\x96\xE6\x9D\xAF20170318041 column=d:comment_list, timestamp=1494234324426, value=[{"content": "\x5Cu6ca1\x5Cu60f3\x5Cu5230\x5Cu4f60\x5Cu662f\x5Cu8fd9\x5Cu6837\x5Cu7684\x
                                                 5Cu7b11\x5Cu7b11", "area": "\x5Cu65e5\x5Cu672c", "name": "ZJK\x5Cu7684JK\x5Cu4e0d\x5Cu662f\x5Cu5973\x5Cu9ad8\x5Cu751f\x5Cu7684JK", "time": "20
                                                 17-03-18"}, {"content": "+\x5Cu7231\x5Cu539f\x5Cu6765\x5Cu5c31\x5Cu4e3a\x5Cu7684\x5Cu662f\x5Cu76f8\x5Cu805a\x5Cuff0f\x5Cu4e3a\x5Cu7684\x5Cu662
                                                 f\x5Cu4e0d\x5Cu518d\x5Cu5206\x5Cu79bb\x5Cuff0f\x5Cu82e5\x5Cu6709\x5Cu4e00\x5Cu79cd\x5Cu7231\x5Cu662f\x5Cu6c38\x5Cu4e0d\x5Cu80fd\x5Cu76f8\x5Cu8
                                                 9c1\x5Cuff0f\x5Cu6c38\x5Cu4e0d\x5Cu80fd\x5Cu542f\x5Cu53e3\x5Cuff0f\x5Cu6c38\x5Cu4e0d\x5Cu80fd\x5Cu518d\x5Cu60f3\x5Cu8d77\x5Cuff0f\x5Cu5c31\x5C
                                                 u597d\x5Cu50cf\x5Cu6c38\x5Cu4e0d\x5Cu80fd\x5Cu71c3\x5Cu8d77\x5Cu7684\x5Cu706b\x5Cu79cd\x5Cuff0f\x5Cu5b64\x5Cu72ec\x5Cu5730\x5Cuff0f\x5Cu51dd\x
                                                 5Cu671b\x5Cu7740\x5Cu9ed1\x5Cu6697\x5Cu7684\x5Cu5929\x5Cu7a7a", "area": "\x5Cu6c5f\x5Cu82cf\x5Cu82cf\x5Cu5dde", "name": "\x5Cu7528\x5Cu6237597
                                                 5181560", "time": "2017-03-18"}, {"content": "\x5Cu803f\x5Cu76f4boy@\x5Cu8fde\x5Cu7b11", "area": "\x5Cu9ed1\x5Cu9f99\x5Cu6c5f\x5Cu54c8\x5Cu5c1
                                                 4\x5Cu6ee8", "name": "\x5Cu4e0d\x5Cu60f3\x5Cu8d77\x5Cu5e8a\x5Cu6ef4", "time": "2017-03-18"}, {"content": "\x5Cu68cb\x5Cu624b\x5Cu8fde\x5Cu7b11
                                                 \x5Cu7b11\x5Cu7b11", "area": "\x5Cu6c5f\x5Cu82cf\x5Cu5357\x5Cu4eac", "name": "\x5Cu6cfc\x5Cu58a8\x5Cu5206\x5Cu4e94\x5Cu8272", "time": "2017-03
                                                 -18"}, {"content": "\x5Cu597d\x5Cu840c", "area": "\x5Cu5e7f\x5Cu897f\x5Cu67f3\x5Cu5dde", "name": "\x5Cu4e0d\x5Cu7f9e\x5Cu5c0f\x5Cu9a6c\x5Cu8e4
                                                 4", "time": "2017-03-18"}, {"content": "\x5Cu6d3b\x5Cu6349\x5Cu6781\x5Cu54c1\x5Cu6b63\x5Cu592a\x5Cu4e00\x5Cu679a\x5Cu3002", "area": "\x5Cu65b0
                                                 \x5Cu7586\x5Cu514b\x5Cu62c9\x5Cu739b\x5Cu4f9d", "name": "2172290834", "time": "2017-03-18"}, {"content": "\x5Cu53ef\x5Cu80fd\x5Cu4e5d\x5Cu5c81
                                                 \x5Cu7684\x5Cu65f6\x5Cu5019\x5Cuff0c\x5Cu6211\x5Cu6709\x5Cu4e00\x5Cu6b21\x5Cu4e0a\x5Cu5395\x5Cu6240\x5Cuff0c\x5Cu7279\x5Cu522b\x5Cu597d\x5Cu73
                                                 a9\x5Cuff0c\x5Cu6211\x5Cu7a81\x5Cu7136\x5Cu611f\x5Cu89c9\x5Cuff0c\x5Cu81ea\x5Cu5df1\x5Cu5728\x5Cu601d\x5Cu8003\x5Cu4eba\x5Cu751f\x5Cu4e86\x5Cu
                                                 3002[\x5Cu563b\x5Cu563b]\x5Cu633a\x5Cu9017\x5Cu7684\x5Cuff0c\x5Cu7b11\x5Cu7b11\x5Cu5f97\x5Cu52a0\x5Cu6cb9\x5Cu4e89\x5Cu53d6\x5Cu62ff\x5Cu4e2a\
                                                 x5Cu4e16\x5Cu754c\x5Cu5927\x5Cu8d5b\x5Cu51a0\x5Cu519b\x5Cu4e86\x5Cu3002", "area": "\x5Cu5317\x5Cu4eac", "name": "1828193193", "time": "2017-03
                                                 -18"}]                                                                                                                                        
 \xE5\xAB\x98\xE7\xA5\x96\xE6\x9D\xAF20170318041 column=d:good, timestamp=1494234324426, value=6                                                                                               
 \xE5\xAB\x98\xE7\xA5\x96\xE6\x9D\xAF20170318041 column=d:source, timestamp=1494234324426, value=\xE6\x96\xB0\xE6\xB5\xAA\xE4\xBD\x93\xE8\x82\xB2                                              
 \xE5\xAB\x98\xE7\xA5\x96\xE6\x9D\xAF20170318041 column=d:time, timestamp=1494234324426, value=2017-03-18                                                                                      
 \xE5\xAB\x98\xE7\xA5\x96\xE6\x9D\xAF20170318041 column=d:title, timestamp=1494234324426, value=\xE8\xBF\x9E\xE7\xAC\x91\xEF\xBC\x9A\xE5\x90\x8D\xE4\xBA\xBA\xE6\x88\x98\xE5\xA4\xBA\xE5\x86\xA
                                                 0\xE6\x97\xB6\xE5\xBF\x83\xE5\xA6\x82\xE6\xAD\xA2\xE6\xB0\xB4_\xE6\x89\x8B\xE6\x9C\xBA\xE6\x96\xB0\xE6\xB5\xAA\xE7\xBD\x91                    
 \xE5\xAB\x98\xE7\xA5\x96\xE6\x9D\xAF20170318041 column=d:url, timestamp=1494234324426, value=https://sports.sina.cn/others/qipai/2017-03-18/detail-ifycnpiu8996035.d.html?from=wap&HTTPS=1    
 \xE5\xAB\x98\xE7\xA5\x96\xE6\x9D\xAF20170318041 column=d:web_from, timestamp=1494234324426, value=sports.sina.cn                                                                              
 \xE5\xAB\x98\xE7\xA5\x96\xE6\x9D\xAF20170318041 column=d:web_name, timestamp=1494234324426, value=\xE6\x89\x8B\xE6\x9C\xBA\xE6\x96\xB0\xE6\xB5\xAA\xE7\xBD\x91 '''

keyConv = "examples.pythonconverters.ImmutableBytesWritableToStringConverter"                                                                                                    
      valueConv = "examples.pythonconverters.HBaseResultToStringConverter"                                                                         
      rconf = {"hbase.zookeeper.quorum": host,                                                                                                                       
               "hbase.mapreduce.inputtable": hbase_table}                                                                                                            
      ps_data = sc.newAPIHadoopRDD(                                                                                                                                  
          "org.apache.hadoop.hbase.mapreduce.TableInputFormat",                                                                                                      
          "org.apache.hadoop.hbase.io.ImmutableBytesWritable",                                                                                                       
          "org.apache.hadoop.hbase.client.Result",                                                                                                                   
          keyConverter=rkeyConv,                                                                                                                                     
          valueConverter=rvalueConv,                                                                                                                                 
          conf=rconf                                                                                                                                                 
      )    
    real_data = ps_data.map(lambda x: {x[0]: transform_from_hbase(x[1])})
```

这里没有采用
rkeyConv = "org.apache.spark.examples.pythonconverters.ImmutableBytesWritableToStringConverter"
rvalueConv = "org.apache.spark.examples.pythonconverters.HBaseResultToStringConverter"是因为example里面的Scala采用toStringBinary的格式，然后pyspark提取出来的时候会将'\xaf\x23'等变为u'\\xaf\\x23'这种乱码，原本以json形式存在HBase中的数据会更乱，所以这里就采用了另一个converter，scala源码如下：

```scala
/* file name: pythonConverter.scala
package hbase.pythonconverters

import scala.collection.JavaConverters._
import scala.util.parsing.json.JSONObject

/* Don't use ..python.Convertions*/
import org.apache.spark.api.python.Converter
import org.apache.hadoop.hbase.client.{Put, Result}
import org.apache.hadoop.hbase.io.ImmutableBytesWritable
import org.apache.hadoop.hbase.util.Bytes
import org.apache.hadoop.hbase.KeyValue.Type
import org.apache.hadoop.hbase.CellUtil

/**
 * Implementation of [[org.apache.spark.api.python.Converter]] that converts all 
 * the records in an HBase Result to a String
 */
class HBaseResultToStringConverter extends Converter[Any, String] {
        override def convert(obj: Any): String = { 
    val result = obj.asInstanceOf[Result]
    val output = result.listCells.asScala.map(cell =>
                    Map(
          "row" -> Bytes.toString(CellUtil.cloneRow(cell)),
          "columnFamily" -> Bytes.toString(CellUtil.cloneFamily(cell)),
          "qualifier" -> Bytes.toString(CellUtil.cloneQualifier(cell)),
          "timestamp" -> cell.getTimestamp.toString,
          "type" -> Type.codeToType(cell.getTypeByte).toString,
          "value" -> Bytes.toString(CellUtil.cloneValue(cell))
            
                       )   
    )
    output.map(JSONObject(_).toString()).mkString("\n")

}

/**
 * Implementation of [[org.apache.spark.api.python.Converter]] that converts an
 * ImmutableBytesWritable to a String
 */
class ImmutableBytesWritableToStringConverter extends Converter[Any, String] {
        override def convert(obj: Any): String = {
    val key = obj.asInstanceOf[ImmutableBytesWritable]
    Bytes.toString(key.get())
        }

}

/**
 * Implementation of [[org.apache.spark.api.python.Converter]] that converts a
 * String to an ImmutableBytesWritable
 */
class StringToImmutableBytesWritableConverter extends Converter[Any, ImmutableBytesWritable] {
        override def convert(obj: Any): ImmutableBytesWritable = {
    val bytes = Bytes.toBytes(obj.asInstanceOf[String])
    new ImmutableBytesWritable(bytes)

}

/**
 * Implementation of [[org.apache.spark.api.python.Converter]] that converts a
 * list of Strings to HBase Put
 */
class StringListToPutConverter extends Converter[Any, Put] {
        override def convert(obj: Any): Put = {
    val output = obj.asInstanceOf[java.util.ArrayList[String]].asScala.map(Bytes.toBytes).toArray
    val put = new Put(output(0))
    put.add(output(1), output(2), output(3))

}

```

package hbase.pythonconverters 表示使用时，调用的命名空间为hbase.pythonconverters.
然后采用sbt进行编译：

```
1、文件目录如下：
spark_hbase:
    build.sbt
    src
        main
            scala
                hbase
                    pythonConverter.scala
        test
    project
        build.properties
        plugins.sbt
    target

# build.sbt:
lazy val root = (project in file(".")). 
  settings(              
    name := "spark_hbase",    
    version := "1.0",    
    scalaVersion := "2.10.5", 
    sparkVersion := "1.6.0"   
  )

libraryDependencies ++= Seq(
  "org.apache.hbase" % "hbase-client" % "1.2.0" % "provided",
  "org.apache.hbase" % "hbase-common" % "1.2.0" % "provided",
  "org.apache.hbase" % "hbase-server" % "1.2.0" % "provided"
)

#build.properties
// This file should only contain the version of sbt to use.
sbt.version=0.13.6

# plugins.sbt
// Use sbt-spark-package plugin to assembly package
// You may use this file to add plugin dependencies for sbt.
resolvers += "Spark Packages repo" at "https://dl.bintray.com/spark-packages/maven/"
                         
addSbtPlugin("org.spark-packages" %% "sbt-spark-package" % "0.2.1")

2、安装sbt
$ wget https://dl.bintray.com/sbt/native-packages/sbt/0.13.6/sbt-0.13.6.tgz
$ sudo tar -zxvf sbt-0.13.6.tgz -C /opt
$ sudo vim /etc/profile
export SBT_HOME=/opt/sbt
export PATH=${PATH}:${SBT_HOME}/bin

/*第一次执行时，会下载一些文件包(~/.ivy2)，然后才能正常使用，要确保联网了*/
$ sbt -version

3、打包成jar
# sbt 参数如下：
# package:创建一个 jar 包其中包含 src/main/resources 和编译 src/main/scala 或 src/main/java 目录的 class 文件
# assembly:创建一个jar包，会包涵源文件需要的所有依赖包，如果这里使用assembly命令，有可能导致jar包包涵一个与spark不同的slf4j包，导致warn报错
$ sbt package

# 打包的结果是在target文件下出现一个spark_hbase_2.10-1.0.jar的包

4、运行
# 在hbase新建一个名为test的数据库，列族名为data
$ habse shell
create 'test', 'data'

# 新建一个spark任务文件
$ vim spark_hbase.py
# coding:utf-8
# usage:$ spark-submit --master local --driver-class-path <the path to spark_hbase.jar>/spark_hbase_2.1.0-1.0.jar  spark_hbase.py

import json

from pyspark import SparkContext, SparkConf

def transform_from_hbase(data):
    if not data:
        return {}
    data = data.split('\n')
    tmp = {}
    for x in data:
        td = json.loads(x)
        key = '%s:%s' % (td['columnFamily'], td['qualifier'])
        tmp[key] = td['value']
    return tmp
        
if __name__ == '__main__':
    conf = SparkConf().setAppName('group_url')
    sc = SparkContext(conf=conf)
    
    # save to hbase
    host = '192.168.10.23' # The ip to your hbase
    hbase_table = 'test'  # The hbase database name
    conf = {"hbase.zookeeper.quorum": host,
            "hbase.mapred.outputtable": hbase_table,
            "mapreduce.outputformat.class": "org.apache.hadoop.hbase.mapreduce.TableOutputFormat",
            "mapreduce.job.output.key.class": "org.apache.hadoop.hbase.io.ImmutableBytesWritable",
            "mapreduce.job.output.value.class": "org.apache.hadoop.io.Writable"
            } 
    keyConv = "hbase.pythonconverters.StringToImmutableBytesWritableConverter"
    valueConv = "hbase.pythonconverters.StringListToPutConverter"

# sparkrow1:行关键字; 'd':列族;  'line_name':列名; 'value1':值
    dl= [['sparkrow1', 'd', 'line_name', 'value1'], ['row2', 'd', 'web_name', u'人民网']]
    sc.parallelize(dl).map(lambda x: (x[0], x))\
        .saveAsNewAPIHadoopDataset(conf, keyConv, valueConv)
    
    # read from hbase
    rkeyConv = "hbase.pythonconverters.ImmutableBytesWritableToStringConverter"
    rvalueConv = "hbase.pythonconverters.HBaseResultToStringConverter"
    rconf = {"hbase.zookeeper.quorum": host,
             "hbase.mapreduce.inputtable": hbase_table}
    ps_data = sc.newAPIHadoopRDD(
        "org.apache.hadoop.hbase.mapreduce.TableInputFormat",
        "org.apache.hadoop.hbase.io.ImmutableBytesWritable",
        "org.apache.hadoop.hbase.client.Result",
        keyConverter=rkeyConv,
        valueConverter=rvalueConv,
        conf=rconf
    )

real_data = ps_data.map(lambda x: {x[0]: transform_from_hbase(x[1])})
    for v in real_data.collect():
        print v

sc.stop()

# 运行任务
$ spark-submit --master local[*] --driver-class-path <the path to spark_hbase.jar>/spark_hbase_2.1.0-1.0.jar  spark_hbase.py

# ps:可以在.../spark/conf/spark-defaults.sh配置
spark.executor.extraClassPath=/<the local path to your jar>spark_hbase_2.10-1.0.jar:/<another jar>another.jar
spark.driver.extraClassPath=/<the local path to your jar>spark_hbase_2.10-1.0.jar:/<another jar>another.jar

# 这样提交任务时就可以不用指定--driver-class-path了

```

在读数据时，可以根据需要设置scan参数：

```python
hbase.mapreduce.scan.row.start
hbase.mapreduce.scan.row.stop
hbase.mapreduce.scan.column.family
hbase.mapreduce.scan.columns
hbase.mapreduce.scan.timestamp
hbase.mapreduce.scan.timerange.start
hbase.mapreduce.scan.timerange.end
hbase.mapreduce.scan.maxversions
hbase.mapreduce.scan.cacheblocks
hbase.mapreduce.scan.cachedrows
hbase.mapreduce.scan.batchsize
```

hbase scan注意点：http://blog.csdn.net/lxf310/article/details/19563831
converter编写参考资料：http://www.codesec.net/view/208034.html
sbt入门：http://wiki.jikexueyuan.com/project/sbt-getting-started/
spark构建Scala程序：http://www.cnblogs.com/missmzt/p/6002858.html
spark源码：https://github.com/apache/spark/blob/branch-1.6/examples/src/main/scala/org/apache/spark/examples/pythonconverters/HBaseConverters.scala
spark

pyspark中操作HBase数据以及sbt简单用法

附件: