本篇文章以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 举报,一经查实,本站将立刻删除。