【Flink实战】玩转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

• 保存视频订单到Mysql

  CREATE TABLE `video_order` (
    `id` int(11) unsigned NOT NULL AUTO_INCREMENT,
    `user_id` int(11) DEFAULT NULL,
    `money` int(11) DEFAULT NULL,
    `title` varchar(32) DEFAULT NULL,
    `trade_no` varchar(64) DEFAULT NULL,
    `create_time` date DEFAULT NULL,
    PRIMARY KEY (`id`)
  ) ENGINE=InnoDB DEFAULT CHARSET=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();
      }
  }