flink 自定义 Source，模拟 1000 条订单数据，每条数据间隔 1 秒钟， 包括（订单编号，用户编号，消费金额，消费时间），将数据写入文件。

Apache Flink 的自定义源（Custom Source）允许你在流处理应用程序中创建新的数据生成源，比如模拟数据。为了模拟 1000 条订单数据，我们可以编写一个实现了 ` RichParallelSourceFunction` 或者 `DataStreamSource` 接口的类。这里是一个简单的 Python 示例，使用 Flink 的 Java API，因为我们需要定时生成并序列化数据：

import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.source.RichParallelSourceFunction;

public class OrderSimulator extends RichParallelSourceFunction<String> {
    private int count = 0;
    private long lastGeneratedTime = System.currentTimeMillis();

    @Override
    public void run(SourceContext<String> ctx) throws Exception {
        while (true) {
            if (count >= 1000) { // 如果达到设定的数据量就结束
                cancel信号源函数；
                break;
            }

            // 模拟数据生成，每秒一条订单
            Thread.sleep(1000);
            String order = "Order_" + count + ",User_1,100.0," + generateTimestamp();
            ctx.collect(order);

            count++;
        }
    }

    @Override
    public void cancel() {
        // 取消源任务
        // 这里可以添加更详细的清理工作
    }

    private String generateTimestamp() {
        return Long.toString(System.currentTimeMillis() - lastGeneratedTime);
    }

    @Override
    public void emitWatermark(Watermark mark) {
        // 设置水印，这里是基于当前时间戳的简单示例
        long currentTime = System.currentTimeMillis();
        if (currentTime > mark.getTimestamp()) {
            ctx.emitWatermark(new Watermark(currentTime));
        }
    }
}

public static void main(String[] args) throws Exception {
    final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

    DataStream<String> orders = env.addSource(new OrderSimulator());

    // 写入文件，假设 sink.txt 是目标文件路径
    orders.print().setParallelism(1).writeAsText("sink.txt");

    env.execute("Flink Order Simulation");
}

这个例子中，我们创建了一个订单数据源，它每秒生成一条订单记录，并将其发送到下游的打印操作，最后将所有数据追加到文件 `sink.txt` 中。