如何在 jdbc 数据源中使用 dbtable 选项的子查询？

Question

我想使用 Spark 处理来自 JDBC 源的一些数据。但首先，我不想从 JDBC 读取原始表，而是想在 JDBC 端运行一些查询来过滤列和连接表，并将查询结果作为表加载到 Spark SQL 中。

以下加载原始 JDBC 表的语法适用于我：

df_table1 = sqlContext.read.format('jdbc').options(
    url="jdbc:mysql://foo.com:3306",
    dbtable="mydb.table1",
    user="me",
    password="******",
    driver="com.mysql.jdbc.Driver" # mysql JDBC driver 5.1.41
).load() 
df_table1.show() # succeeded

根据 Spark documentation（我使用的是 PySpark 1.6.3）：

dbtable：应读取的 JDBC 表。请注意，任何有效的 可以在 SQL 查询的 FROM 子句中使用。例如，而不是 完整的表，您还可以在括号中使用子查询。

所以只是为了实验，我尝试了一些简单的方法：

df_table1 = sqlContext.read.format('jdbc').options(
    url="jdbc:mysql://foo.com:3306",
    dbtable="(SELECT * FROM mydb.table1) AS table1",
    user="me",
    password="******",
    driver="com.mysql.jdbc.Driver"
).load() # failed

它抛出了以下异常：

com.mysql.jdbc.exceptions.jdbc4.MySQLSyntaxErrorException: You have an error in your SQL syntax; check the manual that corresponds to your MySQL server version for the right syntax to use near 'table1 WHERE 1=0' at line 1

我还尝试了一些其他的语法变体（添加/删除括号、删除“as”子句、切换大小写等），但没有任何运气。那么正确的语法是什么？在哪里可以找到更详细的语法文档？此外，错误消息中这个奇怪的“WHERE 1=0”是从哪里来的？谢谢！

Answer 1

在 Python 上使用 Spark 2.2 连接到 MySQL (5.7.19) 时，我可以在使用 table="(SELECT * FROM a_table) AS my_table" 时运行以下命令。

from pyspark.sql import SparkSession

my_spark = SparkSession \
    .builder.appName("myApp") \
    .config("jars", "/usr/local/spark-2.2.2-bin-hadoop2.7/jars/mysql-connector-java-5.1.45-bin.jar") \
    .getOrCreate()

my_df = my_spark.read.jdbc(
    url="jdbc:mysql://my_host:3306/my_db",
    table="(SELECT * FROM a_table) AS my_table",
    properties={'user': 'my_username', 'password': 'my_password'}
)

my_df.head(20)

Answer 2

table = "(SELECT id, person, manager, CAST(tdate AS CHAR) AS tdate, CAST(start AS   CHAR) AS start, CAST(end AS CHAR) as end, CAST(duration AS CHAR) AS duration FROM EmployeeTimes) AS EmployeeTimes",

spark = get_spark_session()
df = spark.read.format("jdbc"). \
    options(url=ip,
            driver='com.mysql.jdbc.Driver',
            dbtable=table,
            user=username,
            password=password).load()
return df

Spark JDBC 与 MYSQL 时间戳不兼容，我遇到了很多麻烦。诀窍是在 JDBC 接触它们之前将所有时间戳或持续时间值转换为字符串。只需将您的值转换为字符串，它就会起作用。

注意：您还必须使用 AS 为查询提供别名以使其正常工作。

Answer 3

我认为这可能是 Spark SQL 中的一个错误。

似乎this 或this line 给了你错误。两者都使用 Scala 字符串插值将table 替换为dbtable。

s"SELECT * FROM $table WHERE 1=0"

您可以从您遇到的错误中找到table1 WHERE 1=0，因为上述模式将变为：

SELECT * FROM (select * from table1) as table1 WHERE 1=0

哪个看起来不正确。

确实有一种特定于 MySQL 的方言 - MySQLDialect - 用 its own 覆盖 getTableExistsQuery：

override def getTableExistsQuery(table: String): String = {
  s"SELECT 1 FROM $table LIMIT 1"
}

所以我敢打赌，另一种方法getSchemaQuery 是错误的根源。考虑到您使用 Spark 1.6.3 而该方法具有 @Since("2.1.0") 标记，这不太可能很难。

我强烈建议您查看 MySQL 数据库的日志，看看执行了什么查询会导致错误消息。

Answer 4

对于使用 Spark SQL 中的 sql 查询从 JDBC 源读取数据，您可以尝试以下操作：

val df_table1 = sqlContext.read.format("jdbc").options(Map(
    ("url" -> "jdbc:postgresql://localhost:5432/mydb"),
    ("dbtable" -> "(select * from table1) as table1"),
    ("user" -> "me"),
    ("password" -> "******"),
    ("driver" -> "org.postgresql.Driver"))
).load()

我使用 PostgreSQL 进行了尝试。可以根据MySQL修改。