使用 phoenix 连接器将 Spark 数据帧写入 Hbase

Posted 2023-04-17

【中文标题】使用 phoenix 连接器将 Spark 数据帧写入 Hbase

【英文标题】：Write Spark dataframe to Hbase using phoenix connector

【发布时间】：2017-09-20 07:15:12

【问题描述】：

我有指向 Hbase 表的 Hive 表。我有一个 spark 作业，它创建具有等于 hbase 表的模式的数据集。我正在使用以下命令将此数据帧保存到 hbase 表中。

sql.write().format("org.apache.phoenix.spark")
    .mode(SaveMode.Overwrite).option("table", targetTable)
    .option("zkUrl", "localhost:2181:/hbase-unsecure)
    .insertInto(targetTable);

执行时出现以下错误..

java.lang.NullPointerException
at org.apache.phoenix.hive.PhoenixStorageHandler.configureJobProperties(PhoenixStorageHandler.java:185)
at org.apache.phoenix.hive.PhoenixStorageHandler.configureOutputJobProperties(PhoenixStorageHandler.java:130)
at org.apache.spark.sql.hive.HiveTableUtil$.configureJobPropertiesForStorageHandler(TableReader.scala:324)
at org.apache.spark.sql.hive.SparkHiveWriterContainer.<init>(hiveWriterContainers.scala:67)
at org.apache.spark.sql.hive.execution.InsertIntoHiveTable.sideEffectResult$lzycompute(InsertIntoHiveTable.scala:226)
at org.apache.spark.sql.hive.execution.InsertIntoHiveTable.sideEffectResult(InsertIntoHiveTable.scala:142)
at org.apache.spark.sql.hive.execution.InsertIntoHiveTable.doExecute(InsertIntoHiveTable.scala:310)
at org.apache.spark.sql.execution.SparkPlan$$anonfun$execute$1.apply(SparkPlan.scala:115)
at org.apache.spark.sql.execution.SparkPlan$$anonfun$execute$1.apply(SparkPlan.scala:115)
at org.apache.spark.sql.execution.SparkPlan$$anonfun$executeQuery$1.apply(SparkPlan.scala:136)
at org.apache.spark.rdd.RDDOperationScope$.withScope(RDDOperationScope.scala:151)
at org.apache.spark.sql.execution.SparkPlan.executeQuery(SparkPlan.scala:133)
at org.apache.spark.sql.execution.SparkPlan.execute(SparkPlan.scala:114)
at org.apache.spark.sql.execution.QueryExecution.toRdd$lzycompute(QueryExecution.scala:86)
at org.apache.spark.sql.execution.QueryExecution.toRdd(QueryExecution.scala:86)
at org.apache.spark.sql.DataFrameWriter.insertInto(DataFrameWriter.scala:259)
at org.apache.spark.sql.DataFrameWriter.insertInto(DataFrameWriter.scala:239)
at com.lti.unitrax.data.load.IncrementalHiveTableLoadUnitraxMain.fullDataLoad(IncrementalHiveTableLoadUnitraxMain.java:166)
at com.lti.unitrax.data.load.TestDataLoad.main(TestDataLoad.java:38)
at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
at sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62)
at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
at java.lang.reflect.Method.invoke(Method.java:498)
at org.apache.spark.deploy.yarn.ApplicationMaster$$anon$2.run(ApplicationMaster.scala:627)

非常感谢任何帮助。

我正在使用 spark2 和 HDP 集群。

【问题讨论】：

有一个 Spark HBase 连接器，你知道的，对吧？

你有什么解决办法吗？

你的数据框中到底有什么？有空值吗？您是否尝试过在这里使用 pagerank 示例？ phoenix.apache.org/phoenix_spark.html

【参考方案1】：

我知道我在游戏中迟到了，但看到了这篇文章。以为我的回答可以帮助某人。

这是我在实施过程中的做法

df = 数据帧 zookeeperURL = 集群的 zookeeper URL _tgtTable = 您要写入数据的表

df.write.format("org.apache.phoenix.spark")
  .mode(org.apache.spark.sql.SaveMode.Overwrite)
  .options(collection.immutable.Map("zkUrl" -> zookeeperURL, "table" -> _tgtTable)) 
  .save()