Control insert database with dataset

[Dulce Morim]

Hello,

I'm trying catch a BatchUpdateException when insert DataSet using a method output. Because, I need control if insert a duplicate key. How I can do this?



[2018-06-18 22:18:56,419] INFO DataSink (org.apache.flink.api.java.io.jdbc.JDBCOutputFormat@64aad6db) (1/1) (00a77c9e18f893cde9c62a3c9ca5c471) switched from RUNNING to FAILED. (org.apache.flink.runtime.executiongraph.ExecutionGraph)
java.lang.IllegalArgumentException: writeRecord() failed
        at org.apache.flink.api.java.io.jdbc.JDBCOutputFormat.writeRecord(JDBCOutputFormat.java:209)
        at org.apache.flink.api.java.io.jdbc.JDBCOutputFormat.writeRecord(JDBCOutputFormat.java:41)
        at org.apache.flink.runtime.operators.DataSinkTask.invoke(DataSinkTask.java:194)
        at org.apache.flink.runtime.taskmanager.Task.run(Task.java:702)
        at java.lang.Thread.run(Thread.java:748)
Caused by: java.sql.BatchUpdateException: Violation of PRIMARY KEY constraint 'TEST_PK'. Cannot insert duplicate key in object 'TEST'. The duplicate key value is (371855553).
        at com.microsoft.sqlserver.jdbc.SQLServerPreparedStatement.executeBatch(SQLServerPreparedStatement.java:2303)
        at org.apache.flink.api.java.io.jdbc.JDBCOutputFormat.writeRecord(JDBCOutputFormat.java:205)
        ... 4 more


Only have a generic exception:
org.apache.flink.runtime.client.JobExecutionException: Job execution failed.


Thanks,
Dulce Morim

[Fabian Hueske-2]

Hi Dulce,

This functionality is not supported by the JDBCOutputFormat.

Some database systems (AFAIK, MySQL) support Upsert writes, i.e., writes that insert if the primary key is not present or update the row if the PK exists. Not sure if that would meet your requirements.

If you don't want to go with Upserts or if your DBMS doesn't support them in a single statement, you'd have to implement your own output format.

You can of course use the JDBCOutputFormat as a starting point and adjust it to your needs.

Best, Fabian

2018-06-18 23:38 GMT+02:00 Dulce Morim <[hidden email]>:

Hello,

I'm trying catch a BatchUpdateException when insert DataSet using a method output. Because, I need control if insert a duplicate key. How I can do this?



[2018-06-18 22:18:56,419] INFO DataSink (org.apache.flink.api.java.io.jdbc.JDBCOutputFormat@64aad6db) (1/1) (00a77c9e18f893cde9c62a3c9ca5c471) switched from RUNNING to FAILED. (org.apache.flink.runtime.executiongraph.ExecutionGraph)
java.lang.IllegalArgumentException: writeRecord() failed
        at org.apache.flink.api.java.io.jdbc.JDBCOutputFormat.writeRecord(JDBCOutputFormat.java:209)
        at org.apache.flink.api.java.io.jdbc.JDBCOutputFormat.writeRecord(JDBCOutputFormat.java:41)
        at org.apache.flink.runtime.operators.DataSinkTask.invoke(DataSinkTask.java:194)
        at org.apache.flink.runtime.taskmanager.Task.run(Task.java:702)
        at java.lang.Thread.run(Thread.java:748)
Caused by: java.sql.BatchUpdateException: Violation of PRIMARY KEY constraint 'TEST_PK'. Cannot insert duplicate key in object 'TEST'. The duplicate key value is (371855553).
        at com.microsoft.sqlserver.jdbc.SQLServerPreparedStatement.executeBatch(SQLServerPreparedStatement.java:2303)
        at org.apache.flink.api.java.io.jdbc.JDBCOutputFormat.writeRecord(JDBCOutputFormat.java:205)
        ... 4 more


Only have a generic exception:
org.apache.flink.runtime.client.JobExecutionException: Job execution failed.


Thanks,
Dulce Morim