Apache Beam 2.3 Java版自定义Sink创建方法咨询

嘿，这个问题我刚好熟悉！你提到的旧版com.google.cloud.dataflow.sdk.io.Sink类确实已经不存在了——这是因为Apache Beam从Google Dataflow独立出来后，对IO模块做了全面的API重构，包结构和核心类都有明显变化。

现在怎么实现自定义Sink？ #

在Beam 2.x Java SDK中，你需要使用org.apache.beam.sdk.io包下的相关类来实现自定义Sink，主要有两种适配不同场景的方式：

1. 实现底层Sink接口（通用场景）

如果你的写入逻辑不依赖文件系统（比如写入数据库、外部API等），可以直接实现Sink接口，它包含三个核心协作组件：

• Sink：定义Sink的基础配置，负责创建对应的WriteOperation
• WriteOperation：管理写入操作的生命周期，创建具体的Writer实例
• Writer：实际执行元素写入的逻辑，包含写入、关闭、收尾等方法

举个简单的示例代码：

import org.apache.beam.sdk.io.Sink;
import org.apache.beam.sdk.options.PipelineOptions;
import org.apache.beam.sdk.transforms.PTransform;
import org.apache.beam.sdk.values.PCollection;
import org.apache.beam.sdk.values.PDone;

// 自定义Sink核心实现
public class MyCustomSink implements Sink<String> {
    @Override
    public WriteOperation<String, ?, ?> createWriteOperation(PipelineOptions options) {
        return new MyWriteOperation(options);
    }

    // 自定义写入操作管理器
    private static class MyWriteOperation extends WriteOperation<String, Void, Void> {
        public MyWriteOperation(PipelineOptions options) {
            super(options);
        }

        @Override
        public Writer<String, Void> createWriter(PipelineOptions options) throws Exception {
            return new MyWriter();
        }

        @Override
        public Void finalizeWrite(Void writerResult, PipelineOptions options) throws Exception {
            // 写入完成后的收尾操作，比如关闭数据库连接、提交事务
            return null;
        }
    }

    // 自定义写入器
    private static class MyWriter extends Writer<String, Void> {
        @Override
        public void write(String element) throws Exception {
            // 替换成你的实际写入逻辑，比如写入数据库或外部服务
            System.out.println("Writing element: " + element);
        }

        @Override
        public Void close() throws Exception {
            // 关闭资源，比如流、连接
            return null;
        }
    }
}

// 包装成PTransform方便在Pipeline中调用
public class MySinkTransform extends PTransform<PCollection<String>, PDone> {
    @Override
    public PDone expand(PCollection<String> input) {
        return input.apply(Sink.write(new MyCustomSink()));
    }
}

2. 继承FileBasedSink（文件写入场景）

如果你的Sink是基于文件系统的写入，Beam提供了FileBasedSink抽象类，它已经封装了文件分片、命名、容错等通用逻辑，你只需要重写核心的写入方法即可：

import org.apache.beam.sdk.io.FileBasedSink;
import org.apache.beam.sdk.io.fs.ResourceId;
import org.apache.beam.sdk.options.PipelineOptions;

public class MyFileSink extends FileBasedSink<String> {
    public MyFileSink(ResourceId baseOutputPath) {
        super(baseOutputPath);
    }

    @Override
    public Writer<String> createWriter(PipelineOptions options) throws Exception {
        return new MyFileWriter(this, options);
    }

    private static class MyFileWriter extends FileBasedSink.Writer<String> {
        public MyFileWriter(FileBasedSink<String> sink, PipelineOptions options) throws Exception {
            super(sink, options);
        }

        @Override
        public void write(String element) throws Exception {
            // 写入文件的逻辑，比如按行写入文本
            getOutputStream().write((element + "\n").getBytes());
        }
    }
}

额外提示 #

如果你的写入逻辑非常简单，也可以直接用ParDo来快速实现，但实现Sink接口更符合Beam的IO规范，能更好地利用Beam的容错机制、并行写入优化等特性。

内容的提问来源于stack exchange，提问作者function