Bigquery - Spark的Simba JDBC错误

Question

我正在尝试使用Simba JDBC驱动程序(bigquery jdbc 4.2)从apache spark 2.2.0中查询bigquery

这是我的代码：

import spark.implicits._
import org.apache.spark.sql.SQLContext
val sqlcontext = new org.apache.spark.sql.SQLContext(sc)
val pushdown_query = "(select * from mydb.mytable) AS T"
val dataframe_bq = sqlcontext.read.format("jdbc").option("url","jdbc:bigquery://https://www.googleapis.com/bigquery/v2:443;ProjectId=xxxx;OAuthType=0;OAuthServiceAcctEmail==xxxxxxx@xxxxxx.iam.gserviceaccount.com;OAuthPvtKeyPath=/path/file.json;").option("driver", "com.simba.googlebigquery.jdbc42.Driver").option("dbtable", pushdown_query).load()
dataframe_bq.show()

这是错误：

org.apache.spark.SparkException: Job aborted due to stage failure: Task 0 in stage 3.0 failed 1 times, most recent failure: Lost task 0.0 in stage 3.0 (TID 3, localhost, executor driver): java.sql.SQLDataException: 

[Simba][JDBC](10140) Error converting value to Timestamp.

任何数据类型都可能发生转换错误，例如：

[Simba][JDBC](10140) Error converting value to Long.

Spark 2.x让每个columnName都被双引号引用！

例如，从用户中选择UserID、UserName

将被操作以从用户选择"UserID“、"UserName”

数据帧将如下所示：

+------+--------+ 
|UserID|UserName| 
+------+--------+ 
|UserID|UserName| 
|UserID|UserName| 
|UserID|UserName| 
+------+--------+ 

建议的解决方案是编写自定义的spark JDBC方言，但我还不能让它正常运行。

如果你能告诉我如何解决这个问题，我将不胜感激。提前谢谢。

Answer 1

面临着同样的问题。

如果您需要使用Spark JDBC，除了使用/实现自定义JDBC方言之外别无选择。你可以看看https://github.com/trK54Ylmz/spark-bigquery，它对我来说很有效，只做了很小的改动。

如果JDBC不是必需的，您也可以尝试可用的连接器之一:从Spotify (目前处于维护模式)、Google的原生connector for Dataproc (不幸的是，非常有限)或this第三方版本(看起来很有希望)。

Answer 2

为了解决这个问题，我不得不编辑Apache Spark 2.1版的源代码，并使用以下过程重新编译它来处理这个问题：

1)先决条件

 Maven 3.3.9 or newer and Java 8+ 

2)下载spark源代码，在我的例子中是2.1版

3)更新源代码，防止Spark在列名中添加双引号！，搜索以下代码，并进行相应替换。

源代码参考文件：https://github.com/apache/spark/blob/branch-2.1/sql/core/src/main/scala/org/apache/spark/sql/jdbc/JdbcDialects.scala#L90

编辑此函数：

def quoteIdentifier(colName: String): String = {
    s""""$colName""""
}

让它像这样：

def quoteIdentifier(colName: String): String = {
    return colName
}

4)用maven构建

  Install maven 3.3.9 or above and java 8 and then build it:

    ./dev/make-distribution.sh --name custom-spark --pip --r --tgz -Psparkr -Phadoop-2.7 -Phive -Phive-thriftserver -Pmesos -Pyarn

希望这会对你有所帮助。

Answer 3

当我使用Spark的Simba jdbc驱动时，选项UseNativeQuery=0解决了以下问题：

[Simba][JDBC](10140) Error converting value to Timestamp

因此，希望我不需要开发任何特定的方言。