Apache Phoenix for Spark不工作

Question

我无法通过Spark(2.1.0)连接到Phoenix(4.10)，基于Phoenix网站上的“使用Data Source API作为DataFrame加载”示例。我使用的是lastet(Phoenix4.10)和Hbase 1.2.5。我可以通过Phoenix（sqlline客户端）在Hbase中创建一个表。Spark内返回的错误如下：

scala> val df = sqlContext.load("org.apache.phoenix.spark",Map("table" -> "test", "zkUrl" -> "localhost:2181"))

warning: there was one deprecation warning; re-run with -deprecation for details
java.sql.SQLException: org.apache.hadoop.ipc.RemoteException(org.apache.hadoop.hbase.TableExistsException): SYSTEM.MUTEX
at org.apache.phoenix.query.ConnectionQueryServicesImpl$12.call(ConnectionQueryServicesImpl.java:2465)
at org.apache.phoenix.query.ConnectionQueryServicesImpl$12.call(ConnectionQueryServicesImpl.java:2382)
at org.apache.phoenix.util.PhoenixContextExecutor.call(PhoenixContextExecutor.java:76)
at org.apache.phoenix.query.ConnectionQueryServicesImpl.init(ConnectionQueryServicesImpl.java:2382)
at org.apache.phoenix.jdbc.PhoenixDriver.getConnectionQueryServices(PhoenixDriver.java:255)
at org.apache.phoenix.jdbc.PhoenixEmbeddedDriver.createConnection(PhoenixEmbeddedDriver.java:149)
at org.apache.phoenix.jdbc.PhoenixDriver.connect(PhoenixDriver.java:221)
at java.sql.DriverManager.getConnection(DriverManager.java:664)
at java.sql.DriverManager.getConnection(DriverManager.java:208)
at org.apache.phoenix.mapreduce.util.ConnectionUtil.getConnection(ConnectionUtil.java:98)
at org.apache.phoenix.mapreduce.util.ConnectionUtil.getInputConnection(ConnectionUtil.java:57)
at org.apache.phoenix.mapreduce.util.ConnectionUtil.getInputConnection(ConnectionUtil.java:45)
at org.apache.phoenix.mapreduce.util.PhoenixConfigurationUtil.getSelectColumnMetadataList(PhoenixConfigurationUtil.java:292)
at org.apache.phoenix.spark.PhoenixRDD.toDataFrame(PhoenixRDD.scala:118)
at org.apache.phoenix.spark.PhoenixRelation.schema(PhoenixRelation.scala:60)
at org.apache.spark.sql.execution.datasources.LogicalRelation.<init>(LogicalRelation.scala:40)
at org.apache.spark.sql.SparkSession.baseRelationToDataFrame(SparkSession.scala:389)
at org.apache.spark.sql.DataFrameReader.load(DataFrameReader.scala:146)
at org.apache.spark.sql.DataFrameReader.load(DataFrameReader.scala:125)
at org.apache.spark.sql.SQLContext.load(SQLContext.scala:965)
... 50 elided
Caused by: org.apache.hadoop.ipc.RemoteException: SYSTEM.MUTEX
at org.apache.hadoop.hbase.master.procedure.CreateTableProcedure.prepareCreate(CreateTableProcedure.java:285)
at org.apache.hadoop.hbase.master.procedure.CreateTableProcedure.executeFromState(CreateTableProcedure.java:106)
at org.apache.hadoop.hbase.master.procedure.CreateTableProcedure.executeFromState(CreateTableProcedure.java:58)
at org.apache.hadoop.hbase.procedure2.StateMachineProcedure.execute(StateMachineProcedure.java:119)
at org.apache.hadoop.hbase.procedure2.Procedure.doExecute(Procedure.java:498)
at org.apache.hadoop.hbase.procedure2.ProcedureExecutor.execProcedure(ProcedureExecutor.java:1147)
at org.apache.hadoop.hbase.procedure2.ProcedureExecutor.execLoop(ProcedureExecutor.java:942)
at org.apache.hadoop.hbase.procedure2.ProcedureExecutor.execLoop(ProcedureExecutor.java:895)
at org.apache.hadoop.hbase.procedure2.ProcedureExecutor.access$400(ProcedureExecutor.java:77)
at org.apache.hadoop.hbase.procedure2.ProcedureExecutor$2.run(ProcedureExecutor.java:497)

更新1：如果通过HBase删除System.Mutex表，它可以正常工作。

更新2：删除System.Mutex表后，只要通过sqlcontext.load（）与Phoenix建立连接，就会重新创建该表，这意味着在加载另一个表的瞬间，或者即使重新加载同一个表，在试图重新创建System.Mutex表时也会引发相同的异常。

更新3：如果在Hbase中没有System.Mutex表的情况下启动，那么对于相同的Spark会话，它可以正常工作，也就是说，您可以连接到任意多个表，但是，如果初始化了另一个Spark会话，则从第二个Spark上下文中抛出相同的异常。

根据https://issues.apache.org/jira/browse/phoenix-3814上的建议（在Spark类路径中包括hbase-client jar)，它仍然会出现相同的异常。

更新4：我最终完成了凤凰项目的自定义构建。解决办法是改变线路号。2427在org.apache.phoenix.query.ConnectionQueryServiceSimpl（phoenix-core）类中添加到，如果（！admin.tableexists(SYSTEM_MUTEX_NAME_BYTES))createSysMutexTable(admin)；。另外，在https://phoenix.apache.org/phoenix_spark.html中给出的数据帧加载示例是不正确的，因为它基于DataFrame类的deprecated/removed save方法，而是需要使用write方法。参见下面的示例：

./bin/spark-shell --master local[4] --deploy-mode client --jars path_to_to/phoenix-4.10.1-HBase-1.2-SNAPSHOT-client.jar
import org.apache.spark.SparkContext
import org.apache.spark.sql.SQLContext
import org.apache.phoenix.spark._
import org.apache.spark.sql.SaveMode

val sqlContext = new SQLContext(sc)

val df = sqlContext.load("org.apache.phoenix.spark",Map("table" -> "name_of_input_table_in_phoenix", "zkUrl" -> "localhost:2181"))
df.write.format("org.apache.phoenix.spark").mode(SaveMode.Overwrite).options(Map("table" -> "name_of_output_table_in_phoenix","zkUrl" -> "localhost:2181")).save()

注意，输出表应该已经存在于Phoenix中，具有正确的模式。请注意，我使用的是自定义构建，因此快照是在客户机jar名称中。

Answer 1

当前的4.10版本似乎有这样一个bug，即在初始化时（当在SQLContext上调用load时），Phoenix客户机试图创建System.Mutex表（org.apache.Phoenix.query.ConnectionQueryServiceSimpl（phoenix-core）类中的createSysMutexTable方法）。但是，如果此表已经存在，则HBase将引发TableExistsException。虽然createSysMutexTable方法会捕获TableAlreadyExists execption，但这与Hbase抛出的execption不同，Hbase的异常是包装的。因此，发生未处理的异常。解决方案是更新代码，并仅在互斥表不存在时调用createSysMutexTable方法。有关完整的解决方案和示例代码，请参见更新4。