Подключение к AWS Redshift с Zeppelin Spark 2.0 и Pyspark

Question

Мне нужно прочитать данные Redshift в dataframes в Цеппелине. За последние несколько месяцев я использовал Spark 2.0 через Zeppelin на AWS, чтобы успешно открывать файлы csv и json S3.

Я использовал, чтобы иметь возможность подключения к RedShift от Цеппелина на AWS ОГО с искрой 1.6.2 (возможно, 1.6.1), используя этот код:

%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: An error occurred while calling o119.jdbc. : java.sql.SQLException: No suitable driver at java.sql.DriverManager.getDriver(DriverManager.java:315) at org.apache.spark.sql.execution.datasources.jdbc.JdbcUtils$$anonfun$2.apply(JdbcUtils.scala:54) at org.apache.spark.sql.execution.datasources.jdbc.JdbcUtils$$anonfun$2.apply(JdbcUtils.scala:54) at scala.Option.getOrElse(Option.scala:121) at org.apache.spark.sql.execution.datasources.jdbc.JdbcUtils$.createConnectionFactory(JdbcUtils.scala:53) at org.apache.spark.sql.execution.datasources.jdbc.JDBCRDD$.resolveTable(JDBCRDD.scala:123) at org.apache.spark.sql.execution.datasources.jdbc.JDBCRelation.(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) (, Py4JJavaError(u'An error occurred while calling o119.jdbc.\n', JavaObject id=o121),)

Я исследовал документацию Спарк 2.0, и нашло следующее:

The JDBC driver class must be visible to the primordial class loader on the client session and on all executors. This is because Java’s DriverManager class does a security check that results in it ignoring all drivers not visible to the primordial class loader when one goes to open a connection. One convenient way to do this is to modify compute_classpath.sh on all worker nodes to include your driver JARs.

Я не знаю, как осуществить это и сделал больше чтения из различных сообщений, некоторые блоги и некоторые посты в StackOverflow и нашел это:

spark.driver.extraClassPath = org.postgresql.Driver

я сделал это на странице настроек Interpreter для Zeppelin, но все же я получаю ту же ошибку.

Я попытался добавить переводчика Postgres, и я не уверен, что сделал все правильно (потому что я не был уверен, поместить его в интерпретатор Spark или интерпретатор Python), и я выбрал интерпретатор Spark. Теперь интерпретатор Postgres также имеет все те же настройки, что и интерпретатор Spark, что может не иметь значения, но все же я получаю ту же ошибку.

В Spark 1.6, я просто не помню, чтобы пройти через все эти неприятности.

В качестве эксперимента я развернул кластер EMR с помощью Spark 1.6.2 и попробовал старый код, который использовался для работы, и получил ту же ошибку, что и выше!

На сайте 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