如何使用Python连接HBase和Spark？

Question

我有一个令人尴尬的并行任务,我使用Spark来分配计算.这些计算是在Python中进行的,我使用PySpark来读取和预处理数据.我的任务的输入数据存储在HBase中.不幸的是,我还没有找到一种令人满意的(即易于使用和可扩展的)方法来使用Python从/向Spark读取/写入HBase数据.

我之前探讨的内容:

• 使用我的Python进程连接happybase.该软件包允许使用HBase的Thrift API从Python连接到HBase.这样,我基本上跳过Spark进行数据读/写,并且错过了潜在的HBase-Spark优化.读取速度似乎相当快,但写入速度很慢.这是目前我最好的解决方案.

• 使用SparkContext newAPIHadoopRDD并saveAsNewAPIHadoopDataset使用HBase的MapReduce接口.这方面的例子曾经包含在Spark代码库中(见这里).但是,这些现在被认为是过时的,有利于HBase的Spark绑定(见这里).我还发现这个方法很慢而且很麻烦(用于阅读,编写工作得很好),例如,返回的字符串newAPIHadoopRDD必须以各种方式进行解析和转换,最终得到我想要的Python对象.它一次只支持一列.

我知道的替代方案:

• 我目前正在使用Cloudera的CDH和5.7.0版本hbase-spark(CDH发行说明和详细的博客文章).该模块(以前称为SparkOnHBase)将正式成为HBase 2.0的一部分.不幸的是,这个奇妙的解决方案似乎只适用于Scala/Java.

• 华为的Spark-SQL-on-HBase/Astro(我看不出两者之间有什么区别......).它看起来并不像我想要的解决方案那样强大且受到良好支持.

Answer 1

我发现其中一位制作者的评论hbase-spark,似乎表明有一种方法可以使用PySpark使用Spark SQL查询HBase.

事实上,这里描述的模式可以应用于使用PySpark使用Spark SQL查询HBase,如下例所示:

from pyspark import SparkContext
from pyspark.sql import SQLContext

sc = SparkContext()
sqlc = SQLContext(sc)

data_source_format = 'org.apache.hadoop.hbase.spark'

df = sc.parallelize([('a', '1.0'), ('b', '2.0')]).toDF(schema=['col0', 'col1'])

# ''.join(string.split()) in order to write a multi-line JSON string here.
catalog = ''.join("""{
    "table":{"namespace":"default", "name":"testtable"},
    "rowkey":"key",
    "columns":{
        "col0":{"cf":"rowkey", "col":"key", "type":"string"},
        "col1":{"cf":"cf", "col":"col1", "type":"string"}
    }
}""".split())


# Writing
df.write\
.options(catalog=catalog)\  # alternatively: .option('catalog', catalog)
.format(data_source_format)\
.save()

# Reading
df = sqlc.read\
.options(catalog=catalog)\
.format(data_source_format)\
.load()

我已经尝试过hbase-spark-1.2.0-cdh5.7.0.jar(由Cloudera分发),但遇到了麻烦(org.apache.hadoop.hbase.spark.DefaultSource does not allow create table as select写作java.util.NoSuchElementException: None.get时,阅读时).事实证明,当前版本的CDH不包括hbase-spark允许Spark SQL-HBase集成的更改.

什么不工作对我来说是shc星火包,发现在这里.我必须对上面的脚本进行的唯一更改是更改:

data_source_format = 'org.apache.spark.sql.execution.datasources.hbase'

以下是我在CDH集群中提交上述脚本的方法,遵循shc自述文件中的示例:

spark-submit --packages com.hortonworks:shc:1.0.0-1.6-s_2.10 --repositories http://repo.hortonworks.com/content/groups/public/ --files /opt/cloudera/parcels/CDH/lib/hbase/conf/hbase-site.xml example.py

shc似乎已经将大部分工作合并到hbase-sparkHBase 的模块中,以便在2.0版本中发布.这样,使用上述模式可以使用Spark SQL查询HBase(有关详细信息,请参阅:https: //hbase.apache.org/book.html#_sparksql_dataframes).上面的例子显示了PySpark用户的样子.

最后,需要注意的是:我上面的示例数据只有字符串.Python数据转换不受支持shc,所以我遇到的问题是整数和浮点数没有出现在HBase中或者带有奇怪的值.