使用 Zeppelin Spark 2.0 和 Pyspark 连接到 AWS Redshift

Question

我需要将 Redshift 数据读入 Zeppelin 中的数据帧。在过去的几个月里，我一直在 AWS 上通过 Zeppelin 使用 Spark 2.0 成功打开 csv 和 json S3 文件。

我曾经能够使用 Spark 1.6.2（可能是 1.6.1）从 AWS EMR 上的 Zeppelin 连接到 Redshift，使用以下代码：

%pyspark

from pyspark.sql import SQLContext, Row
import sys
from pyspark.sql.window import Window
import pyspark.sql.functions as func

#Load the data
aquery = "(SELECT serial_number, min(date_time) min_date_time from schema.table where serial_number in ('abcdefg','1234567') group by serial_number) as minDates"

dfMinDates = sqlContext.read.format('jdbc').options(url='jdbc:postgresql://dadadadaaaredshift.amazonaws.com:5439/idw?tcpKeepAlive=true&ssl=true&sslfactory=org.postgresql.ssl.NonValidatingFactory?user=user&password=password', dbtable=aquery).load()
dfMinDates.show()

它奏效了。那是 2016 年夏天。

从那时起我就不再需要它了，现在 AWS 有了 Spark 2.0。

新语法是

myDF = spark.read.jdbc 像这样：

%pyspark

aquery = "(SELECT serial_number, min(date_time) min_date_time from schema.table where serial_number in ('abcdefg','1234567') group by serial_number) as minDates"

dfMinDates = spark.read.jdbc("jdbc:postgresql://dadadadaaaredshift.amazonaws.com:5439/idw?tcpKeepAlive=true&ssl=true&sslfactory=org.postgresql.ssl.NonValidatingFactory?user=user&password=password", dbtable=aquery).load()
dfMinDates.show()

但我收到此错误：

Py4JJavaError：调用 o119.jdbc 时出错。 ： java.sql.SQLException：没有合适的驱动程序 java.sql.DriverManager.getDriver(DriverManager.java:315) 在 org.apache.spark.sql.execution.datasources.jdbc.JdbcUtils$$anonfun$2.apply(JdbcUtils.scala:54) 在 org.apache.spark.sql.execution.datasources.jdbc.JdbcUtils$$anonfun$2.apply(JdbcUtils.scala:54) 在 scala.Option.getOrElse(Option.scala:121) 在 org.apache.spark.sql.execution.datasources.jdbc.JdbcUtils$.createConnectionFactory(JdbcUtils.scala:53) 在 org.apache.spark.sql.execution.datasources.jdbc.JDBCRDD$.resolveTable(JDBCRDD.scala:123) 在 org.apache.spark.sql.execution.datasources.jdbc.JDBCRelation.(JDBCRelation.scala:117) 在 org.apache.spark.sql.DataFrameReader.jdbc(DataFrameReader.scala:237) 在 org.apache.spark.sql.DataFrameReader.jdbc(DataFrameReader.scala:159) 在 sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method) 在 sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62) 在 sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43) 在 java.lang.reflect.Method.invoke(Method.java:498) 在 py4j.reflection.MethodInvoker.invoke(MethodInvoker.java:237) 在 py4j.reflection.ReflectionEngine.invoke(ReflectionEngine.java:357) 在 py4j.Gateway.invoke(Gateway.java:280) 在 py4j.commands.AbstractCommand.invokeMethod(AbstractCommand.java:128) 在 py4j.commands.CallCommand.execute(CallCommand.java:79) 在 py4j.GatewayConnection.run(GatewayConnection.java:211) 在 java.lang.Thread.run(Thread.java:745) (, Py4JJavaError(u'发生错误 调用 o119.jdbc.\n', JavaObject id=o121), )

我研究了 Spark 2.0 文档，发现：

JDBC 驱动程序类必须对原始类加载器可见 在客户端会话和所有执行者上。这是因为 Java 的 DriverManager 类进行安全检查，导致它忽略 当一个驱动程序运行时，所有驱动程序对原始类加载器都不可见 打开连接。一种方便的方法是修改 所有工作节点上的 compute_classpath.sh 以包含您的驱动程序 JAR。

我不知道如何实现这一点，并在 stackoverflow 中阅读了各种帖子、一些博客和一些帖子，发现了这个：

spark.driver.extraClassPath = org.postgresql.Driver

我在 Zeppelin 的解释器设置页面中执行了此操作，但仍然出现相同的错误。

我尝试添加一个 Postgres 解释器，但我不确定我做对了（因为我不确定是否将它放在 Spark 解释器或 Python 解释器中），我选择了 Spark 解释器。现在 Postgres 解释器也具有与 Spark 解释器相同的所有设置，这可能无关紧要，但我仍然得到相同的错误。

在 Spark 1.6 中，我只是不记得经历过所有这些麻烦。

作为一个实验，我用 Spark 1.6.2 启动了一个 EMR 集群，并尝试了以前可以工作的旧代码，并得到了与上面相同的错误！

Zeppelin 站点包含 Postgres，但它们的信息看起来像代码，而不是如何设置解释器，所以我不知道如何使用它。

我没有想法和参考。

非常感谢任何建议！

Answer 1

您需要使用 Amazon 的 Redshift 特定驱动程序。您可以从这里下载：http://docs.aws.amazon.com/redshift/latest/mgmt/configure-jdbc-connection.html。

但是，如果您使用的是 EMR，它已经就位（/usr/share/aws/redshift/jdbc/RedshiftJDBC41.jar），您只需告诉 Zeppelin 它在哪里。

声明方法如下：AWS Redshift driver in Zeppelin