flink中读取kafka数据设置水位线最大乱序程度为3秒

在flink中读取kafka数据并设置水位线最大乱序程度为3秒，可以使用以下代码：

import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.timestamps.BoundedOutOfOrdernessTimestampExtractor;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer;

import java.util.Properties;

public class KafkaSourceExample {
    public static void main(String[] args) throws Exception {
        // set up the streaming execution environment
        final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        // set up properties for kafka consumer
        Properties props = new Properties();
        props.setProperty("bootstrap.servers", "localhost:9092");
        props.setProperty("group.id", "test");

        // create a kafka consumer source
        FlinkKafkaConsumer<String> kafkaSource = new FlinkKafkaConsumer<>("my-topic", new SimpleStringSchema(), props);

        // set up watermark generator with max out-of-order delay of 3 seconds
        kafkaSource.assignTimestampsAndWatermarks(new BoundedOutOfOrdernessTimestampExtractor<String>(Time.seconds(3)) {
            @Override
            public long extractTimestamp(String element) {
                // extract event timestamp from the input data
                // assuming the timestamp is in the format of "yyyy-MM-dd HH:mm:ss"
                SimpleDateFormat dateFormat = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
                try {
                    Date date = dateFormat.parse(element.split(",")[0]);
                    return date.getTime();
                } catch (ParseException e) {
                    e.printStackTrace();
                    return 0L;
                }
            }
        });

        // add the kafka source to the streaming environment
        env.addSource(kafkaSource).print();

        // execute program
        env.execute("Kafka Source Example");
    }
}

在上述代码中，我们使用 FlinkKafkaConsumer 创建了一个 kafka 消费者源，然后使用 BoundedOutOfOrdernessTimestampExtractor 设置了水位线生成器，最大乱序程度为3秒。在 extractTimestamp 方法中，我们从输入数据中提取了事件时间戳，并将其转换为毫秒级时间戳。最后，我们将 kafka 消费者源添加到流式环境中，并执行程序