如何使用JDBC连接(Py)Spark到Postgres数据库

Question

我已经按照这篇文章的说明从现有的Postgres数据库中读取数据，表名为"objects"，由SQLalchemy中的Objects类定义和创建。在Jupyter笔记本中，我的代码是

from pyspark import SparkContext
from pyspark import SparkConf
from random import random
#spark conf
conf = SparkConf()
conf.setMaster("local[*]")
conf.setAppName('pyspark')
sc = SparkContext(conf=conf)
from pyspark.sql import SQLContext
sqlContext = SQLContext(sc)
properties = {
    "driver": "org.postgresql.Driver"
}
url = 'jdbc:postgresql://PG_USER:PASSWORD@PG_SERVER_IP/db_name'
df = sqlContext.read.jdbc(url=url, table='objects', properties=properties)

最后一行结果如下:

Py4JJavaError: An error occurred while calling o25.jdbc.
: java.lang.NullPointerException
    at org.apache.spark.sql.execution.datasources.jdbc.JDBCRDD$.resolveTable(JDBCRDD.scala:158)
    at org.apache.spark.sql.execution.datasources.jdbc.JDBCRelation.<init>(JDBCRelation.scala:117)
    at org.apache.spark.sql.DataFrameReader.jdbc(DataFrameReader.scala:237)
    at org.apache.spark.sql.DataFrameReader.jdbc(DataFrameReader.scala:159)
    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:237)
    at py4j.reflection.ReflectionEngine.invoke(ReflectionEngine.java:357)
    at py4j.Gateway.invoke(Gateway.java:280)
    at py4j.commands.AbstractCommand.invokeMethod(AbstractCommand.java:128)
    at py4j.commands.CallCommand.execute(CallCommand.java:79)
    at py4j.GatewayConnection.run(GatewayConnection.java:211)
    at java.lang.Thread.run(Thread.java:745)

所以它看起来不能解析表。我如何从这里进行测试以确保我正确地连接到数据库?

Answer 1

名称解析的问题由org.postgresql.util.PSQLException表示，不会导致NPE。问题的根源实际上是一个连接字符串，特别是您提供用户凭据的方式。乍一看，这似乎是一个错误，但如果你正在寻找一个快速的解决方案，你可以使用URL属性:

url = 'jdbc:postgresql://PG_SERVER_IP/db_name?user=PG_USER&password=PASSWORD'

或属性参数:

properties = {
    "user": "PG_USER",
    "password": "PASSWORD",
    "driver": "org.postgresql.Driver"
}