PySpark HBase/Phoenix 集成

Question

我应该将 Phoenix 数据读入 pyspark。

编辑： 我正在使用 Spark HBase 转换器：

这是一个代码sn-p：

port="2181"
host="zookeperserver"
keyConv = "org.apache.spark.examples.pythonconverters.ImmutableBytesWritableToStringConverter"
valueConv = "org.apache.spark.examples.pythonconverters.HBaseResultToStringConverter"
cmdata_conf = {"hbase.zookeeper.property.clientPort":port, "hbase.zookeeper.quorum": host, "hbase.mapreduce.inputtable": "camel", "hbase.mapreduce.scan.columns": "data:a"}
sc.newAPIHadoopRDD("org.apache.hadoop.hbase.mapreduce.TableInputFormat","org.apache.hadoop.hbase.io.ImmutableBytesWritable","org.apache.hadoop.hbase.client.Result",keyConverter=keyConv,valueConverter=valueConv,conf=cmdata_conf)

追溯：

Traceback (most recent call last):
  File "<stdin>", line 1, in <module>
  File "/usr/hdp/2.3.0.0-2557/spark/python/pyspark/context.py", line 547, in newAPIHadoopRDD
    jconf, batchSize)
  File "/usr/hdp/2.3.0.0-2557/spark/python/lib/py4j-0.8.2.1-src.zip/py4j/java_gateway.py", line 538, in __call__
  File "/usr/hdp/2.3.0.0-2557/spark/python/lib/py4j-0.8.2.1-src.zip/py4j/protocol.py", line 300, in get_return_value
py4j.protocol.Py4JJavaError: An error occurred while calling z:org.apache.spark.api.python.PythonRDD.newAPIHadoopRDD.
: java.io.IOException: No table was provided.
    at org.apache.hadoop.hbase.mapreduce.TableInputFormatBase.getSplits(TableInputFormatBase.java:130)

任何帮助将不胜感激。

谢谢！ /蒂娜

Answer 1

推荐使用 spark phoenix 插件。 凤凰火花插件详情请咨询here

环境：使用 AWS EMR 5.10、PySpark 测试

以下是步骤

1. 在phoenix中创建表https://phoenix.apache.org/language/ 打开凤凰壳

   “/usr/lib/phoenix/bin/sqlline.py”

   如果存在表名，则删除表；

   CREATE TABLE 表名（DOMAIN VARCHAR 主键）；

   UPSERT INTO TableName (DOMAIN) VALUES('foo');

2. 下载 spark phoenix 插件 jar 从https://mvnrepository.com/artifact/org.apache.phoenix/phoenix-core/4.11.0-HBase-1.3下载spark phoenix插件jar 你需要 phoenix--HBase--client.jar ，我根据我的 phoenix 和 hbase 版本使用了 phoenix-4.11.0-HBase-1.3-client.jar

3. 在您的 hadoop 主目录中，设置以下变量：

   phoenix_jars=/home/user/apache-phoenix-4.11.0-HBase-1.3-bin/phoenix-4.11.0-HBase-1.3-client.jar

4. 启动 PySpark shell 并在 Driver 和 executer 类路径中添加依赖

   pyspark --jars ${phoenix_jars} --conf spark.executor.extraClassPath=${phoenix_jars}

--创建ZooKeeper URL，替换成你的集群zookeeper quorum，可以从hbase-site.xml查看

emrMaster = "ZooKeeper URL" 

df = sqlContext.read \
.format("org.apache.phoenix.spark") \
.option("table", "TableName") \
.option("zkUrl", emrMaster) \
.load() 

df.show()
df.columns
df.printSchema()
df1=df.replace(['foo'], ['foo1'], 'DOMAIN')
df1.show() 

df1.write \
  .format("org.apache.phoenix.spark") \
  .mode("overwrite") \
  .option("table", "TableName") \
  .option("zkUrl", emrMaster) \
  .save()

Answer 2

有两种方法可以做到这一点：
1) 由于 Phoenix 有 JDBC 层，您可以使用 Spark JdbcRDD 从 Spark https://spark.apache.org/docs/1.3.0/api/java/org/apache/spark/rdd/JdbcRDD.html 中的 Phoenix 读取数据

2) 使用 Spark HBAse 转换器： https://github.com/apache/spark/blob/master/examples/src/main/scala/org/apache/spark/examples/pythonconverters/HBaseConverters.scala https://github.com/apache/spark/tree/master/examples/src/main/python