flink从自定义的偏移量开始消费

在 Flink 中，可以通过实现自定义的 SourceFunction 来控制消费偏移量的起始位置。下面是一个简单的示例：

public class MySource implements SourceFunction<String> {
    
    private String[] data = {"a", "b", "c", "d", "e"};
    private int offset;
    
    @Override
    public void run(SourceContext<String> ctx) throws Exception {
        while (offset < data.length) {
            ctx.collect(data[offset]);
            offset++;
        }
    }
    
    @Override
    public void cancel() {}
}

在上面的示例中，我们通过实现 SourceFunction 接口来生成一组数据。其中，我们维护了一个偏移量 offset，表示当前应该从哪个位置开始消费数据。在 run 方法中，我们不断地从 data 数组中取出数据，并将其发送给 Flink 系统。这样，我们就可以通过控制 offset 的值来指定从哪个位置开始消费数据。

在使用该 SourceFunction 时，我们可以通过 Flink 的 API 来指定起始偏移量。例如，下面的代码将从第三个元素开始消费数据：

StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

DataStreamSource<String> source = env.addSource(new MySource())
        .setParallelism(1)
        .setStartingOffsets(StartingOffsets.fromSpecificOffsets(Collections.singletonMap(new TopicPartition("topic", 0), 2L)));

source.print();

env.execute("My Source Demo");

在上面的代码中，我们通过 `setStartingOffsets` 方法来指定起始偏移量。其中，`fromSpecificOffsets` 方法可以接受一个 Map 对象作为参数，其中的 Key-Value 对应着 Kafka 的 TopicPartition 和对应的偏移量。这样，我们就可以在 Flink 中从指定的位置开始消费数据了。