Flink高性能写入关系型数据库Oracle或者MySql

本篇文章以Oracle为例：

public class SinkOracle extends RichSinkFunction> {
    private Connection connection;
    private PreparedStatement statement;

    // 1,初始化
    @Override
    public void open(Configuration parameters) throws Exception {
        super.open(parameters);
        Class.forName("");
        connection = DriverManager.getConnection("","","");
        String sql = "insert into STREAMING.TRANSACTION VALUES(?,?,?,?)";
        String sql2 = "select * from STREAMING.TRANSACTION";
        statement = connection.prepareStatement(sql);
    }

    // 2,执行
    @Override
    public void invoke(Tuple4 value, Context context) throws Exception {

        System.out.println("value.toString()-------" + value.toString());
        statement.setString(1, value.f0);
        statement.setLong(2, value.f1);
        statement.setString(3, value.f2);
        statement.setDouble(4, value.f3);
        statement.execute();
    }

    // 3,关闭
    @Override
    public void close() throws Exception {
        super.close();
        if (statement != null)
            statement.close();
        if (connection != null)
            connection.close();
    }
}