如何使用 Python 连接 HBase 和 Spark？

Question

我有一个令人尴尬的并行任务，我使用 Spark 来分配计算。这些计算在 Python 中，我使用 PySpark 读取和预处理数据。我的任务的输入数据存储在 HBase 中。不幸的是，我还没有找到使用 Python.

read/write HBase 数据 from/to Spark 的令人满意的（即易于使用和可扩展）方式

我之前探索过的内容：

使用 happybase 从我的 Python 进程中连接。这个包允许使用 HBase 的 Thrift API 从 Python 连接到 HBase。这样，我基本上跳过了数据 reading/writing 的 Spark，并且错过了潜在的 HBase-Spark 优化。读取速度似乎相当快，但写入速度很慢。这是目前我最好的解决方案。
使用 SparkContext 的 newAPIHadoopRDD 和 saveAsNewAPIHadoopDataset，它们利用了 HBase 的 MapReduce 接口。这方面的示例曾经包含在 Spark 代码库中 (see here). However, these are now considered outdated in favor of HBase's Spark bindings (see here)。我还发现这种方法既慢又麻烦（对于读、写效果很好），例如，从 newAPIHadoopRDD 返回的字符串必须以各种方式进行解析和转换，最终得到 Python 我想要的对象。它还一次只支持一列。

我知道的备选方案：

我目前正在使用 Cloudera 的 CDH 和 5.7.0 版提供 hbase-spark (CDH release notes, and a detailed blog post)。该模块（以前称为 SparkOnHBase）将正式成为 HBase 2.0 的一部分。不幸的是，这个绝妙的解决方案似乎只适用于 Scala/Java.
华为的Spark-SQL-on-HBase / Astro（我看不出两者有什么区别。。。）。它看起来不像我希望的解决方案那样强大和得到很好的支持。

Answer 1

我发现 this comment 是由 hbase-spark 的一位开发者提供的，这似乎表明有一种方法可以使用 PySpark 通过 Spark SQL.[=27= 查询 HBase ]

事实上，the pattern described here 可以应用于使用 PySpark 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 不包括允许 Spark SQL-HBase 集成的对 hbase-spark 的更改。

对我有用的是 shc Spark 包，发现 here。我必须对上述脚本进行的唯一更改是更改：

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 的 hbase-spark 模块中，以便在 2.0 版本中发布。这样，就可以使用上述模式对 HBase 进行 Spark SQL 查询（有关详细信息，请参阅：https://hbase.apache.org/book.html#_sparksql_dataframes）。我上面的示例显示了 PySpark 用户的样子。

最后，一个警告：我上面的示例数据只有字符串。 Python shc 不支持数据转换，因此我遇到了整数和浮点数未显示在 HBase 中或具有奇怪值的问题。

如何使用 Python 连接 HBase 和 Spark？

How to connect HBase and Spark using Python?

python

hbase

apache-spark

apache-spark-sql

pyspark