Ошибка удаленного доступа SQL-запроса Scala от GCP к локальной сети - PullRequest
Новая
       75

Ошибка удаленного доступа SQL-запроса Scala от GCP к локальной сети

2 голосов
/ 23 сентября 2019

У меня был следующий код: 

     import org.jooq._ 
     import org.jooq.impl._ 
     import org.jooq.impl.DSL._ 
     import java.sql.DriverManager
     import org.apache.log4j.receivers.db.dialect.SQLDialect

     val session = SparkSession.builder().getOrCreate()
     var df1 = session.emptyDataFrame
     var df2 = session.emptyDataFrame

     val userName = "user"
     val password = "pass"

     val c = DriverManager.getConnection("jdbc:mysql://blah_blah.com", userName, password)

     df1 = sql(s"select * from $db1_name.$tb1_name")
     df2 = c.prepareStatement(s"select * from $db2_name.$tb2_name")

Затем я получил следующую ошибку: 

    found : org.jooq.SQL
    required: org.apache.spark.sql.DataFrame
    (which expands to) 
    org.apache.spark.sql.Dataset[org.apache.spark.sql.Row]
    df1 = sql(s"select * from $db1_name.$tb1_name")
             ^

    found : java.sql.PreparedStatement
    required: org.apache.spark.sql.DataFrame
    (which expands to) 
    org.apache.spark.sql.Dataset[org.apache.spark.sql.Row]
    df2 = c.prepareStatement(s"select * from $db2_name.$tb2_name")

Затем, согласно комментариям, я изменил свой код на:

У меня есть следующий код Scala: 

    val userName = "user"
    val password = "pass"

    val session = SparkSession.builder().getOrCreate()
    var df1 = session.emptyDataFrame
    var df2 = session.emptyDataFrame

    ....
    ....
    df1 = sql(s"select * from $db1_name.$tb1_name")
    df2 = session.read.format("jdbc").
    option("url", "jdbc:mysql://blah_blah.com").
    option("driver", "com.mysql.jdbc.Driver").
    option("useUnicode", "true").
    option("continueBatchOnError","true").
    option("useSSL", "false").
    option("user", userName).
    option("password", password).
    option("dbtable",s"select * from $db2_name.$tb2_name").load()

Я получаю следующие ошибки: 

    The last packet sent successfully to the server was 0 milliseconds 
    ago. The driver has not received any packets from the server.
    at sun.reflect.NativeConstructorAccessorImpl.newInstance0(Native Method)
    at sun.reflect.NativeConstructorAccessorImpl.newInstance(NativeConstructorAccessorImpl.java:62)
    at sun.reflect.DelegatingConstructorAccessorImpl.newInstance(DelegatingConstructorAccessorImpl.java:45)
    at java.lang.reflect.Constructor.newInstance(Constructor.java:423)
    at com.mysql.jdbc.Util.handleNewInstance(Util.java:425)
    at com.mysql.jdbc.SQLError.createCommunicationsException(SQLError.java:989)
    at com.mysql.jdbc.MysqlIO.readPacket(MysqlIO.java:632)
    at com.mysql.jdbc.MysqlIO.doHandshake(MysqlIO.java:1016)
    at com.mysql.jdbc.ConnectionImpl.coreConnect(ConnectionImpl.java:2194)
    at com.mysql.jdbc.ConnectionImpl.connectOneTryOnly(ConnectionImpl.java:2225)
    at com.mysql.jdbc.ConnectionImpl.createNewIO(ConnectionImpl.java:2024)
    at com.mysql.jdbc.ConnectionImpl.<init>(ConnectionImpl.java:779)
    at com.mysql.jdbc.JDBC4Connection.<init>(JDBC4Connection.java:47)
    at sun.reflect.NativeConstructorAccessorImpl.newInstance0(Native Method)
    at sun.reflect.NativeConstructorAccessorImpl.newInstance(NativeConstructorAccessorImpl.java:62)
    at sun.reflect.DelegatingConstructorAccessorImpl.newInstance(DelegatingConstructorAccessorImpl.java:45)
    at java.lang.reflect.Constructor.newInstance(Constructor.java:423)
    at com.mysql.jdbc.Util.handleNewInstance(Util.java:425)
    at com.mysql.jdbc.ConnectionImpl.getInstance(ConnectionImpl.java:389)
    at com.mysql.jdbc.NonRegisteringDriver.connect(NonRegisteringDriver.java:330)
    at org.apache.spark.sql.execution.datasources.jdbc.JdbcUtils$$anonfun$createConnectionFactory$1.apply(JdbcUtils.scala:63)
    at org.apache.spark.sql.execution.datasources.jdbc.JdbcUtils$$anonfun$createConnectionFactory$1.apply(JdbcUtils.scala:54)
    at org.apache.spark.sql.execution.datasources.jdbc.JDBCRDD$.resolveTable(JDBCRDD.scala:56)
    at org.apache.spark.sql.execution.datasources.jdbc.JDBCRelation.<init>(JDBCRelation.scala:115)
    at org.apache.spark.sql.execution.datasources.jdbc.JdbcRelationProvider.createRelation(JdbcRelationProvider.scala:52)
    at org.apache.spark.sql.execution.datasources.DataSource.resolveRelation(DataSource.scala:341)
    at org.apache.spark.sql.DataFrameReader.loadV1Source(DataFrameReader.scala:239)
    at org.apache.spark.sql.DataFrameReader.load(DataFrameReader.scala:227)
    at org.apache.spark.sql.DataFrameReader.load(DataFrameReader.scala:164)
    ... 78 elided
    Caused by: java.io.EOFException: Can not read response from server. 
    Expected to read 4 bytes, read 0 bytes before connection was unexpectedly lost.
    at com.mysql.jdbc.MysqlIO.readFully(MysqlIO.java:3011)
    at com.mysql.jdbc.MysqlIO.readPacket(MysqlIO.java:567)
    ... 100 more

Есть ли какое-либо решение или предложение по поводу этих двух ошибок?

У меня естьпробовал также драйверы postgresql и h2 => org.postgresql.Driver

Но я получаю похожие ошибки (возможно, не точные)

Ответы [ 2 ]

1 голос
/ 24 сентября 2019

Вы можете просто получить DataFrame, прочитав, как показано ниже.Задайте сведения о соединении: 

val jdbcHostname = "some.host.name"
val jdbcDatabase = "some_db"
val driver = "com.mysql.cj.jdbc.Driver" // update driver as needed, In your case it will be `org.postgresql.Driver`
// url to DB
val jdbcUrl = s"jdbc:mysql://$jdbcHostname:3306/$jdbcDatabase"
val username = "someUser"
val password = "somePass"

// create a properties map for your DB connection
val connectionProperties = new Properties()

connectionProperties.put("user", s"${username}")
connectionProperties.put("password", s"${password}")
connectionProperties.setProperty("Driver", driver)

и затем прочитайте из JDBC как: 

// use above created url and connection properties to fetch data
val tableName = "some-table"
val mytable = spark.read.jdbc(jdbcUrl, tableName, connectionProperties)

Spark автоматически считывает схему из таблицы базы данных и сопоставляет ее типы с типами Spark SQL.

Вы можете использовать приведенный выше mytable фрейм данных для выполнения запросов или сохранения данных.

Допустим, вы хотите выбрать столбцы, такие как и сохранить, затем 

// your select query
val selectedDF = mytable.select("c1", "c2")
// now you can save above dataframe
1 голос
/ 24 сентября 2019

Ваша проблема в том, что компилятор scala уже инициализировал var ds1 и ds2 как пустой фрейм данных.Вы должны попытаться прочитать непосредственно из спарк: 

spark.read.format("jdbc")
.option("url", jdbcUrl)
.option("query", "select c1, c2 from t1")
.load()

для получения другой информации вы можете проверить непосредственно на странице Apache Spark

https://spark.apache.org/docs/latest/sql-data-sources-jdbc.html

...