Apache Flink如何实现Sink嵌套？Flink1.14多输出Sink实现疑问

针对你需要在Sink中决定是否写入Kafka和其他数据源的需求，有两种实用方案：

方案一：使用侧输出流（Side Output）分流处理 #

这种方式无需自定义嵌套Sink，通过ProcessFunction将数据按条件分流到不同侧输出流，再分别对接KafkaSink和其他Sink，逻辑清晰且契合Flink数据流设计。

实现步骤： #

• 定义对应Kafka、其他数据源的侧输出流Tag
• 在ProcessFunction中根据业务规则将数据发送到对应侧输出流
• 分别将各侧输出流连接至对应的Sink

代码示例： #

import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.util.Collector;
import org.apache.flink.util.OutputTag;

// 定义侧输出流标识Tag
final OutputTag<YourData> kafkaOutputTag = new OutputTag<YourData>("kafka-output") {};
final OutputTag<YourData> otherOutputTag = new OutputTag<YourData>("other-output") {};

// 主数据流分流处理
SingleOutputStreamOperator<YourData> mainStream = inputStream.process(new ProcessFunction<YourData, YourData>() {
    @Override
    public void processElement(YourData value, Context ctx, Collector<YourData> out) throws Exception {
        // 根据业务条件判断是否发送至Kafka
        if (value.getStatus() == 1) {
            ctx.output(kafkaOutputTag, value);
        }
        // 判断是否发送至其他数据源
        if (value.getType().equals("TYPE_A")) {
            ctx.output(otherOutputTag, value);
        }
    }
});

// 侧输出流对接官方KafkaSink
DataStream<YourData> kafkaStream = mainStream.getSideOutput(kafkaOutputTag);
kafkaStream.sinkTo(kafkaSink);

// 侧输出流对接自定义Sink
DataStream<YourData> otherStream = mainStream.getSideOutput(otherOutputTag);
otherStream.addSink(new YourCustomSink());

方案二：自定义CompositeSinkFunction（封装多Sink逻辑） #

如果必须将所有Sink逻辑整合到一个自定义Sink中，可以实现RichSinkFunction，内部手动管理KafkaProducer和其他数据源客户端（替代官方KafkaSink，因为1.14版本的KafkaSink未公开invoke方法供外部调用）。

实现步骤： #

• 在open方法中初始化KafkaProducer与其他数据源客户端
• 在invoke方法中根据条件决定是否写入对应数据源
• 在close方法中关闭所有资源

代码示例： #

import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.sink.RichSinkFunction;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerRecord;

import java.util.Properties;

public class CompositeSink extends RichSinkFunction<YourData> {
    private KafkaProducer<String, YourData> kafkaProducer;
    private OtherDataSourceClient otherClient;

    @Override
    public void open(Configuration parameters) throws Exception {
        super.open(parameters);
        // 初始化KafkaProducer，配置参考官方KafkaSink参数
        Properties kafkaProps = new Properties();
        kafkaProps.put("bootstrap.servers", "your-kafka-brokers");
        kafkaProps.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
        kafkaProps.put("value.serializer", "your.custom.serializer.YourDataSerializer");
        this.kafkaProducer = new KafkaProducer<>(kafkaProps);

        // 初始化其他数据源客户端
        this.otherClient = new OtherDataSourceClient();
        otherClient.init();
    }

    @Override
    public void invoke(YourData value, Context context) throws Exception {
        // 判断是否写入Kafka
        if (value.getStatus() == 1) {
            ProducerRecord<String, YourData> record = new ProducerRecord<>("your-kafka-topic", value.getId(), value);
            kafkaProducer.send(record);
        }

        // 判断是否写入其他数据源
        if (value.getType().equals("TYPE_A")) {
            otherClient.write(value);
        }
    }

    @Override
    public void close() throws Exception {
        super.close();
        // 关闭资源
        if (kafkaProducer != null) {
            kafkaProducer.flush();
            kafkaProducer.close();
        }
        if (otherClient != null) {
            otherClient.close();
        }
    }
}

// 使用自定义复合Sink
inputStream.addSink(new CompositeSink());

注意事项： #

• 方案二中手动使用KafkaProducer需自行处理序列化、重试、事务等逻辑，不如官方KafkaSink封装完善；若需Exactly-Once语义，需额外配置事务参数
• 优先推荐侧输出流方案，它复用了官方KafkaSink的全部特性，且逻辑解耦、便于维护

内容的提问来源于stack exchange，提问作者Sid-Ant