Connecting to Cloud SQL from Dataflow Job
I am trying to use JdbcIO with Apache Beam 2.0 (Java) to connect to a Cloud SQL instance from Dataflow in the same project.
I am getting the following error:
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.)
-
According to the documentation, the dataflow service account *@dataflow-service-producer-prod.iam.gserviceaccount.com should have access to all resources within the same project if it has been granted Editor permissions.
-
When I run the same Dataflow job with DirectRunner everything works fine.
This is the code I am using:
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));
}
}));
EDIT:
Using the following off-ray approach on another dataflow job seems to work fine with DataflowRunner, which tells me the database might not be the problem.
java.sql.Connection connection = DriverManager.getConnection(JDBC_URL, JDBC_USER, JDBC_PW);
source to share
I think this approach might work better, try com.mysql.jdbc.GoogleDriver and use the maven dependencies listed here.
https://cloud.google.com/appengine/docs/standard/java/cloud-sql/#Java_Connect_to_your_database
Related question: Where do I find and download this jar file com.mysql.jdbc.GoogleDriver?
source to share
Hi this worked for me the way it did. Also, I removed the name and password methods from the db config method and my pipeline settings look like below
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));
}
}));
Hope this helps
source to share
Following these instructions on how to connect to Cloud SQL from Java:
https://cloud.google.com/sql/docs/mysql/connect-external-app#java
I managed to get it to work.
This is what the code looks like (you must replace MYDBNAME, MYSQLINSTANCE, USER and PASSWORD with their values.
Heads up : MYSQLINSTANCE format is project: zone: instancename.
And I'm using a custom class (Customer) to store values for each row instead of key-value pairs.
p.apply(JdbcIO. <Customer> read()
.withDataSourceConfiguration(
JdbcIO.DataSourceConfiguration.create(
"com.mysql.jdbc.Driver",
"jdbc:mysql://google/MYDBNAME?cloudSqlInstance=MYSQLINSTANCE&socketFactory=com.google.cloud.sql.mysql.SocketFactory&user=USER&password=PASSWORD&useUnicode=true&characterEncoding=UTF-8"
)
)
.withQuery( "SELECT CustomerId, Name, Location, Email FROM Customers" )
.withCoder( AvroCoder.of(Customer.class) )
.withRowMapper(
new JdbcIO.RowMapper < Customer > ()
{
@Override
public Customer mapRow(java.sql.ResultSet resultSet) throws Exception
{
final Logger LOG = LoggerFactory.getLogger(CloudSqlToBq.class);
LOG.info(resultSet.getString(2));
Customer customer = new Customer(resultSet.getInt(1), resultSet.getString(2), resultSet.getString(3), resultSet.getString(3));
return customer;
}
}
)
);
Hope this helps.
source to share