從Dataflow作業連接到雲端SQL作業

我很努力地使用JdbcIO和Apache Beam 2.0（Java）在同一個項目中連接到Dataflow的Cloud SQL實例。從Dataflow作業連接到雲端SQL作業

，我發現了以下錯誤：

java.sql.SQLException: Cannot create PoolableConnectionFactory (Communications link failure 

The last packet sent successfully to the server was 0 milliseconds ago. The driver has not received any packets from the server.)

根據文檔數據流服務帳戶*@dataflow-service-producer-prod.iam.gserviceaccount.com應該有訪問所有如果他擁有「編輯者」權限，則可以在同一個項目中找到資源。
當我使用DirectRunner運行相同的Dataflow作業時，一切正常。

這是我使用的代碼：

private static String JDBC_URL = "jdbc:mysql://myip:3306/mydb?verifyServerCertificate=false&useSSL=true"; 

PCollection < KV < String, Double >> exchangeRates = p.apply(JdbcIO. < KV < String, Double >> read() 
.withDataSourceConfiguration(JdbcIO.DataSourceConfiguration.create("com.mysql.jdbc.Driver", JDBC_URL) 
    .withUsername(JDBC_USER).withPassword(JDBC_PW)) 
.withQuery(
    "SELECT CurrencyCode, ExchangeRate FROM mydb.mytable") 
.withCoder(KvCoder.of(StringUtf8Coder.of(), DoubleCoder.of())) 
.withRowMapper(new JdbcIO.RowMapper < KV < String, Double >>() { 
    public KV < String, Double > mapRow(ResultSet resultSet) throws Exception { 
    return KV.of(resultSet.getString(1), resultSet.getDouble(2)); 
    } 
}));

編輯：

使用其他數據流任務中梁的外部設置方法似乎很好地工作DataflowRunner告訴我認爲數據庫可能不是問題。

java.sql.Connection connection = DriverManager.getConnection(JDBC_URL, JDBC_USER, JDBC_PW);

來源

2017-06-22 Jimmy

我覺得這個方法可以更好地工作，請嘗試com.mysql.jdbc.GoogleDriver，並使用這裏列出的Maven依賴。

https://cloud.google.com/appengine/docs/standard/java/cloud-sql/#Java_Connect_to_your_database

來源

2017-06-27 23:29:24

嘿@亞歷克斯·阿馬託，不幸的是，似乎沒有與gcp數據流一起工作，因爲我得到一個「java.sql.SQLException：無法加載JDBC驅動程序類'com.mysql.jdbc.GoogleDriver'」錯誤，即使這兩個maven依賴關係都已添加。 – Jimmy

喜它的方式ü沒有it.Additionaly我刪除從數據庫配置方法withusername和密碼的方法和我的流水線配置工作對我來說看起來像下面

PCollection < KV < Double, Double >> exchangeRates = p.apply(JdbcIO. < KV < Double, Double >> read() 
    .withDataSourceConfiguration(JdbcIO.DataSourceConfiguration.create("com.mysql.jdbc.Driver", "jdbc:mysql://ip:3306/dbname?user=root&password=root&useUnicode=true&characterEncoding=UTF-8") 
      ) 
    .withQuery(
     "SELECT PERIOD_YEAR, PERIOD_YEAR FROM SALE") 
    .withCoder(KvCoder.of(DoubleCoder.of(), DoubleCoder.of())) 
    .withRowMapper(new JdbcIO.RowMapper < KV < Double, Double >>() { 
     @Override 
     public KV<Double, Double> mapRow(java.sql.ResultSet resultSet) throws Exception { 
     LOG.info(resultSet.getDouble(1)+ "Came"); 
      return KV.of(resultSet.getDouble(1), resultSet.getDouble(2)); 
     } 
    }));

希望這將有助於

來源

2017-08-02 11:13:02

從Dataflow作業連接到雲端SQL作業

回答

相關問題