flink的流数据中的时间戳转为时间格式

在Flink中，流数据中的时间戳可以通过AssignerWithPeriodicWatermarks或AssignerWithPunctuatedWatermarks进行转换为时间格式。

使用AssignerWithPeriodicWatermarks：

public class TimestampToTimeAssigner implements AssignerWithPeriodicWatermarks<MyEvent> {

    private final long maxOutOfOrderness = 3000; // 最大乱序时间为3秒
    private long currentMaxTimestamp;

    @Override
    public long extractTimestamp(MyEvent element, long previousElementTimestamp) {
        long timestamp = element.getTimestamp();
        currentMaxTimestamp = Math.max(timestamp, currentMaxTimestamp);
        return timestamp;
    }

    @Override
    public Watermark getCurrentWatermark() {
        return new Watermark(currentMaxTimestamp - maxOutOfOrderness);
    }
}

然后，在Flink的流处理程序中使用该AssignerWithPeriodicWatermarks：

DataStream<MyEvent> stream = ... // 从数据源获取流数据
stream.assignTimestampsAndWatermarks(new TimestampToTimeAssigner())
    .map(...); // 使用时间格式的时间戳进行处理

使用AssignerWithPunctuatedWatermarks：

public class TimestampToTimeAssigner implements AssignerWithPunctuatedWatermarks<MyEvent> {

    private final long maxOutOfOrderness = 3000; // 最大乱序时间为3秒

    @Override
    public long extractTimestamp(MyEvent element, long previousElementTimestamp) {
        long timestamp = element.getTimestamp();
        return timestamp;
    }

    @Override
    public Watermark checkAndGetNextWatermark(MyEvent lastElement, long extractedTimestamp) {
        return new Watermark(extractedTimestamp - maxOutOfOrderness);
    }
}

然后，在Flink的流处理程序中使用该AssignerWithPunctuatedWatermarks：

DataStream<MyEvent> stream = ... // 从数据源获取流数据
stream.assignTimestampsAndWatermarks(new TimestampToTimeAssigner())
    .map(...); // 使用时间格式的时间戳进行处理

以上是两种常用的将流数据中的时间戳转换为时间格式的方法。根据具体需求，选择合适的方法进行使用