请教下，自定义 flink source，我想整个流处理完一批数据，要怎么控制数据的下发?

https://developer.aliyun.com/article/1385874?spm=a2c6h.13148508.setting.14.45724f0eiiHbUd，此回答整理自钉群“【③群】Apache Flink China社区”

要实现自定义 Flink Source，可以通过继承 RichSourceFunction 类来实现。在 RichSourceFunction 中，可以重写 run(SourceContext<T> ctx) 方法来控制数据的下发。

具体来说，可以在 run(SourceContext<T> ctx) 方法中使用一个循环来处理数据流，每次处理完一批数据后，使用 ctx.collect() 方法将结果发送给下游算子，然后等待一段时间再下发下一批数据。可以使用 Thread.sleep() 方法来模拟等待时间。

以下是一个简单的示例代码：

public class MyCustomSource extends RichSourceFunction<String> {
    private List<String> dataList;
    private int index;

    @Override
    public void open(Configuration parameters) throws Exception {
        super.open(parameters);
        dataList = new ArrayList<>(); // 初始化数据列表
        // 加载数据到 dataList 中
    }

    @Override
    public void close() throws Exception {
        super.close();
        // 关闭资源等操作
    }

    @Override
    public void run(SourceContext<String> ctx) throws Exception {
        while (true) {
            // 处理当前批次的数据
            for (int i = index; i < dataList.size(); i++) {
                String data = dataList.get(i);
                // 处理数据，例如过滤、转换等操作
                // ...
                // 将处理后的数据发送给下游算子
                ctx.collect(data);
            }
            // 等待一段时间后下发下一批数据
            Thread.sleep(1000L); // 等待1秒钟
            index = index + dataList.size(); // 更新下标值，以便下次处理下一批数据
        }
    }
}

在上面的代码中，MyCustomSource 类继承了 RichSourceFunction 类，并实现了 open()、close()、run() 三个方法。在 run() 方法中，使用了一个无限循环来处理数据流，每次处理完一批数据后，使用 ctx.collect() 方法将结果发送给下游算子，然后等待一段时间再下发下一批数据。

在Flink中，自定义Source需要实现RichSourceFunction接口，并重写run()方法。在这个方法中，你可以控制数据的下发。具体来说，你可以在run()方法中使用outputCollector来发送数据，并在发送完一批数据后调用outputWatermark()方法来发出水印，表示这一批次的数据已经处理完毕。

以下是一个简单的示例：

import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.typeinfo.BasicTypeInfo;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.source.SourceFunction;
import org.apache.flink.streaming.api.watermark.Watermark;
import org.apache.flink.util.Collector;

public class CustomSource implements SourceFunction<String> {
    private volatile boolean isRunning = true;

    @Override
    public void run(SourceContext<String> ctx) throws Exception {
        while (isRunning) {
            // 生成数据
            String data = generateData();
            // 发送数据
            ctx.collect(data);
            // 发送水印
            ctx.outputWatermark(new Watermark(System.currentTimeMillis()));
        }
    }

    @Override
    public void cancel() {
        isRunning = false;
    }

    private String generateData() {
        // 在这里生成你的数据
        return "data";
    }
}

然后，你可以在Flink的流处理程序中使用这个自定义的Source：

StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
DataStream<String> dataStream = env.addSource(new CustomSource());
dataStream.map(new MapFunction<String, String>() {
    @Override
    public String map(String value) throws Exception {
        // 在这里处理你的数据
        return value;
    }
}).print();
env.execute("Custom Source Example");

在这个示例中，CustomSource会一直运行，每次生成一条数据并发送到下一个Operator。当所有数据都发送完毕后，它会停止运行。