微信公众号搜"智元新知"关注
微信扫一扫可直接关注哦!

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

本篇文章以Oracle为例:

public class SinkOracle extends RichSinkFunction<Tuple4<String, Long, String, Double>> {
    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<String, Long, String, Double> 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();
    }
}

版权声明:本文内容由互联网用户自发贡献,该文观点与技术仅代表作者本人。本站仅提供信息存储空间服务,不拥有所有权,不承担相关法律责任。如发现本站有涉嫌侵权/违法违规的内容, 请发送邮件至 dio@foxmail.com 举报,一经查实,本站将立刻删除。

相关推荐