使用Spark JDBC和Avatica从Apache Druid中的表读取记录

问题描述

我试图在Spark中创建一个数据框，其中将包含Apache Druid中一个表的所有记录，而我正在使用JDBC进行此操作。 Druid似乎正在使用Calcite-Avatica JDBC驱动程序（提到了here）。

    df = spark.read.format('jdbc').option('url','jdbc:avatica:remote:url=http://172.31.5.20:8082/druid/v2/sql/avatica/').option('driver','org.apache.calcite.avatica.remote.Driver').option('dbtable','mytable').load()

但是我遇到了以下错误

Py4JJavaError: An error occurred while calling o456.load.
: java.sql.sqlException: While closing connection
    at org.apache.calcite.avatica.Helper.createException(Helper.java:39)
    at org.apache.calcite.avatica.AvaticaConnection.close(AvaticaConnection.java:156)
    at org.apache.spark.sql.execution.datasources.jdbc.JDBCRDD$.resolveTable(JDBCRDD.scala:70)
    at org.apache.spark.sql.execution.datasources.jdbc.JDBCRelation.<init>(JDBCRelation.scala:115)
    at org.apache.spark.sql.execution.datasources.jdbc.JdbcRelationProvider.createRelation(JdbcRelationProvider.scala:52)
    at org.apache.spark.sql.execution.datasources.DataSource.resolveRelation(DataSource.scala:341)
    at org.apache.spark.sql.DataFrameReader.loadV1Source(DataFrameReader.scala:239)
    at org.apache.spark.sql.DataFrameReader.load(DataFrameReader.scala:227)
    at org.apache.spark.sql.DataFrameReader.load(DataFrameReader.scala:164)
    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 py4j.reflection.MethodInvoker.invoke(MethodInvoker.java:244)
    at py4j.reflection.ReflectionEngine.invoke(ReflectionEngine.java:357)
    at py4j.Gateway.invoke(Gateway.java:282)
    at py4j.commands.AbstractCommand.invokeMethod(AbstractCommand.java:132)
    at py4j.commands.CallCommand.execute(CallCommand.java:79)
    at py4j.GatewayConnection.run(GatewayConnection.java:238)
    at java.lang.Thread.run(Thread.java:748)
Caused by: java.lang.RuntimeException: com.fasterxml.jackson.databind.exc.UnrecognizedPropertyException: Unrecognized field "rpcMetadata" (class org.apache.calcite.avatica.remote.Service$CloseConnectionResponse),not marked as ignorable (0 kNown properties: ])
 at [Source: (String)"{"response":"closeConnection","rpcMetadata":{"response":"rpcMetadata","serverAddress":"ip-172-31-5-234.ap-southeast-1.compute.internal:8082"}}
"; line: 1,column: 46] (through reference chain: org.apache.calcite.avatica.remote.Service$CloseConnectionResponse["rpcMetadata"])
    at org.apache.calcite.avatica.remote.JsonService.handle(JsonService.java:142)
    at org.apache.calcite.avatica.remote.JsonService.apply(JsonService.java:229)
    at org.apache.calcite.avatica.remote.RemoteMeta.closeConnection(RemoteMeta.java:78)
    at org.apache.calcite.avatica.AvaticaConnection.close(AvaticaConnection.java:153)
    ... 18 more
Caused by: com.fasterxml.jackson.databind.exc.UnrecognizedPropertyException: Unrecognized field "rpcMetadata" (class org.apache.calcite.avatica.remote.Service$CloseConnectionResponse),column: 46] (through reference chain: org.apache.calcite.avatica.remote.Service$CloseConnectionResponse["rpcMetadata"])
    at com.fasterxml.jackson.databind.exc.UnrecognizedPropertyException.from(UnrecognizedPropertyException.java:61)
    at com.fasterxml.jackson.databind.DeserializationContext.handleUnkNownProperty(DeserializationContext.java:823)
    at com.fasterxml.jackson.databind.deser.std.StdDeserializer.handleUnkNownProperty(StdDeserializer.java:1153)
    at com.fasterxml.jackson.databind.deser.BeanDeserializerBase.handleUnkNownProperty(BeanDeserializerBase.java:1589)
    at com.fasterxml.jackson.databind.deser.BeanDeserializerBase.handleUnkNownVanilla(BeanDeserializerBase.java:1567)
    at com.fasterxml.jackson.databind.deser.BeanDeserializer.vanillaDeserialize(BeanDeserializer.java:294)
    at com.fasterxml.jackson.databind.deser.BeanDeserializer._deserializeOther(BeanDeserializer.java:189)
    at com.fasterxml.jackson.databind.deser.BeanDeserializer.deserialize(BeanDeserializer.java:161)
    at com.fasterxml.jackson.databind.jsontype.impl.AsPropertyTypeDeserializer._deserializeTypedForId(AsPropertyTypeDeserializer.java:130)
    at com.fasterxml.jackson.databind.jsontype.impl.AsPropertyTypeDeserializer.deserializeTypedFromObject(AsPropertyTypeDeserializer.java:97)
    at com.fasterxml.jackson.databind.deser.BeanDeserializerBase.deserializeWithType(BeanDeserializerBase.java:1178)
    at com.fasterxml.jackson.databind.deser.impl.TypeWrappedDeserializer.deserialize(TypeWrappedDeserializer.java:68)
    at com.fasterxml.jackson.databind.ObjectMapper._readMapAndClose(ObjectMapper.java:4014)
    at com.fasterxml.jackson.databind.ObjectMapper.readValue(ObjectMapper.java:3005)
    at org.apache.calcite.avatica.remote.JsonService.decode(JsonService.java:131)
    at org.apache.calcite.avatica.remote.JsonService.apply(JsonService.java:227)
    ... 20 more

有人知道这可能是什么原因吗，我该如何解决？看来这与Avatica驱动程序有关，似乎它获得了带有无法识别字段的json对象

我正在使用驱动程序org.apache.calcite.avatica:avatica-core:1.17.0，并将jar文件添加到我的spark.jars属性中。我正在使用Druid 0.19.0和Spark2。

编辑：我检查了Avatica JDBC框架的源代码，并注解为@JsonCreator的构造函数期望对json对象中名为rpcMetadata的属性进行反序列化。源代码为here。

解决方法

使用下面的库，然后重试。

"org.apache.calcite.avatica" % "avatica" % "1.8.0"