【Flink实战】玩转Flink里面核心的Sink Operator实战

🚀 作者 ：“大数据小禅”

🚀 文章简介 ：玩转Flink里面核心的Sink Operator实战

🚀 欢迎小伙伴们 点赞👍、收藏⭐、留言💬

目录导航

Flink Sink Operator简介

• 在Flink中，Sink Operator（也称为Sink Function或Sink）是指负责将DataStream或DataSet的数据发送到外部存储或外部系统的操作符。Sink Operator是Flink的数据输出端，它的作用是将处理过的数据写入目标位置，如数据库、文件系统、消息队列等。
• Sink Operator通过将数据传输到外部系统来完成最终的数据存储、展示或其他类型的处理。它可以将数据单个地或批量地发送到目标系统，具体取决于Sink操作符的实现。例如，可以将数据写入关系型数据库、NoSQL数据库、消息队列、文件系统等。
• 在Flink中，可以使用预定义的Sink操作符，如addSink()方法，或自定义Sink函数来实现数据的输出。预定义的Sink操作符可以满足一般的输出需求，而自定义Sink函数可以根据具体的业务逻辑实现特定的输出操作。
• 自定义Sink函数需要实现SinkFunction接口或RichSinkFunction抽象类，并重写其中的方法。这些方法包括open()、invoke()和close()等，用于初始化和管理连接，以及处理数据发送等操作。

使用Sink Operator时，需要考虑以下几个方面：

• 目标系统的可用性和容错性：保证目标系统的可用性，并确保在故障发生时能够进行重试或恢复。
• 写入的一致性：根据需求选择适当的写入一致性级别，如精确一次（exactly-once）或最少一次（at-least-once）语义。
• 并行度和性能：根据目标系统的特性和可用资源，设置合适的并行度以提高任务并行处理和整体性能。

Flink 核心知识 Sink Operator速览

• Flink编程模型
• Sink 输出源 - 预定义 - print- writeAsText （过期）- 自定义 - SinkFunction- RichSinkFunction - Rich相关的api更丰富，多了Open、Close方法，用于初始化连接等- flink官方提供 Bundle Connector - kafka、ES 等- Apache Bahir - kafka、ES、Redis等

Flink 自定义的Sink 连接Mysql存储商品订单案例实战

• 自定义- SinkFunction- RichSinkFunction - Rich相关的api更丰富，多了Open、Close方法，用于初始化连接等
• Flink连接mysql的几种方式（都需要加jdbc驱动）- 方式一：自带flink-connector-jdbc 需要加依赖包<dependency><groupId>org.apache.flink</groupId><artifactId>flink-connector-jdbc_2.12</artifactId><version>1.12.0</version></dependency>- 方式二：自定义sink
• 保存视频订单到MysqlCREATETABLE`video_order`(`id`int(11)unsignedNOTNULLAUTO_INCREMENT,`user_id`int(11)DEFAULTNULL,`money`int(11)DEFAULTNULL,`title`varchar(32)DEFAULTNULL,`trade_no`varchar(64)DEFAULTNULL,`create_time`dateDEFAULTNULL,PRIMARYKEY(`id`))ENGINE=InnoDBDEFAULTCHARSET=utf8mb4;- 添加jdbc依赖<dependency><groupId>mysql</groupId><artifactId>mysql-connector-java</artifactId><version>8.0.25</version></dependency>- 编码public class MysqlSink extends RichSinkFunction<VideoOrder> { private Connection conn = null; private PreparedStatement ps = null; @Override public void open(Configuration parameters) throws Exception { conn = DriverManager.getConnection("jdbc:mysql://localhost:3306/xd_order?useUnicode=true&characterEncoding=utf8&allowMultiQueries=true&serverTimezone=Asia/Shanghai", "root", "xdclass.net"); //url user passwd String sql = "INSERT INTO `video_order` (`user_id`, `money`, `title`, `trade_no`, `create_time`) VALUES(?,?,?,?,?);"; ps = conn.prepareStatement(sql); } @Override public void close() throws Exception { if (conn != null) { conn.close(); } if (ps != null) { ps.close(); } } @Override public void invoke(VideoOrder videoOrder, Context context) throws Exception { //给ps中的?设置具体值 ps.setInt(1,videoOrder.getUserId()); ps.setInt(2,videoOrder.getMoney()); ps.setString(3,videoOrder.getTitle()); ps.setString(4,videoOrder.getTradeNo()); ps.setDate(5,new Date(videoOrder.getCreateTime().getTime())); ps.executeUpdate(); }}