什么是Sink

Sink（接收器）是Flink数据处理流水线的末端，负责将计算结果输出到外部存储系统或下游处理系统。在Flink的编程模型中，Sink是DataStream API中的一个转换操作，它接收DataStream并将数据写入指定的外部系统。

2. Sink的分类

Flink的Sink连接器可以分为以下几类：

• 内置Sink：如print()、printToErr()等用于调试的内置输出
• 文件系统Sink：支持写入本地文件系统、HDFS等
• 消息队列Sink：如Kafka、RabbitMQ等
• 数据库Sink：如JDBC、Elasticsearch等
• 自定义Sink：通过实现SinkFunction接口自定义输出逻辑

3. 输出语义保证

Flink为Sink提供了三种输出语义保证：

• 最多一次（At-most-once）：数据可能丢失，但不会重复
• 至少一次（At-least-once）：数据不会丢失，但可能重复
• 精确一次（Exactly-once）：数据既不会丢失，也不会重复

这些语义保证与Flink的检查点（Checkpoint）机制密切相关，我们将在后面详细讨论。

二、环境准备与依赖配置

1. 版本说明

• Flink：1.20.1
• JDK：17+
• Gradle：8.3+
• 外部系统：Kafka 3.4.0、Elasticsearch 7.17.0、MySQL 8.0

2. 核心依赖

dependencies {
    // Flink核心依赖
    implementation 'org.apache.flink:flink_core:1.20.1'
    implementation 'org.apache.flink:flink-streaming-java:1.20.1'
    implementation 'org.apache.flink:flink-clients:1.20.1'
    
    // Kafka Connector
    implementation 'org.apache.flink:flink-connector-kafka:3.4.0-1.20'
    
    // Elasticsearch Connector
    implementation 'org.apache.flink:flink-connector-elasticsearch7:3.1.0-1.20'
    
    // JDBC Connector
    implementation 'org.apache.flink:flink-connector-jdbc:3.3.0-1.20'
    implementation 'mysql:mysql-connector-java:8.0.33'
    
    // FileSystem Connector
    implementation 'org.apache.flink:flink-connector-files:1.20.1'

}

三、基础Sink操作

1. 内置调试Sink

Flink提供了一些内置的Sink用于开发和调试阶段：

import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

public class BasicSinkDemo {
    public static void main(String[] args) throws Exception {
        // 创建执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        
        // 创建数据源
        DataStream<String> stream = env.fromElements("Hello", "Flink", "Sink");
        
        // 打印到标准输出
        stream.print("StandardOutput");
        
        // 打印到标准错误输出
        stream.printToErr("ErrorOutput");
        
        // 执行作业
        env.execute("Basic Sink Demo");
    }
}

2. 文件系统Sink

Flink支持将数据写入本地文件系统、HDFS等。下面是一个写入本地文件系统的示例：

package com.cn.daimajiangxin.flink.sink;

import org.apache.flink.api.common.serialization.SimpleStringEncoder;
import org.apache.flink.configuration.MemorySize;
import org.apache.flink.connector.file.sink.FileSink;
import org.apache.flink.core.fs.Path;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.sink.filesystem.RollingPolicy;
import org.apache.flink.streaming.api.functions.sink.filesystem.rollingpolicies.DefaultRollingPolicy;

import java.time.Duration;

public class FileSystemSinkDemo {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        DataStream<Object> stream = env.fromData("Hello", "Flink", "FileSystem", "Sink");
        RollingPolicy<Object, String> rollingPolicy = DefaultRollingPolicy.<Object, String>builder()
                .withRolloverInterval(Duration.ofMinutes(15))
                .withInactivityInterval(Duration.ofMinutes(5))
                .withMaxPartSize(MemorySize.ofMebiBytes(64))
                .build();

        // 创建文件系统Sink
        FileSink<Object> sink = FileSink
                .forRowFormat(new Path("file:///tmp/flink-output"), new SimpleStringEncoder<>())
                .withRollingPolicy(rollingPolicy)
                .build();
        // 添加Sink
        stream.sinkTo(sink);
        env.execute("File System Sink Demo");
    }
}

四、高级Sink连接器

1. Kafka Sink

Kafka是实时数据处理中常用的消息队列，Flink提供了强大的Kafka Sink支持：

import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.connector.kafka.sink.KafkaRecordSerializationSchema;
import org.apache.flink.connector.kafka.sink.KafkaSink;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

import java.util.Properties;

public class KafkaSinkDemo {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        
        // 开启检查点以支持Exactly-Once语义
        env.enableCheckpointing(5000);
        
        DataStream<String> stream = env.fromElements("Hello Kafka", "Flink to Kafka", "Data Pipeline");
        
        // Kafka配置
        Properties props = new Properties();
        props.setProperty("bootstrap.servers", "localhost:9092");
        
        // 创建Kafka Sink
        KafkaSink<String> sink = KafkaSink.<String>
                builder()
                .setKafkaProducerConfig(props)
                .setRecordSerializer(KafkaRecordSerializationSchema.builder()
                        .setTopic("flink-output-topic")
                        .setValueSerializationSchema(new SimpleStringSchema())
                        .build())
                .build();
        
        // 添加Sink
        stream.sinkTo(sink);
        
        env.execute("Kafka Sink Demo");
    }
}

kafka消息队列消息：