flink过滤日志数据中的脏数据非JSON格式的字符串并提取ts字段作为最大时间戳生成水位线最大乱序程度为5秒

可以使用Flink的filter算子过滤掉非JSON格式的字符串，然后使用Flink的map算子提取ts字段作为时间戳，并使用Flink的assignTimestampsAndWatermarks算子生成水位线。

下面是一个示例代码：

DataStream<String> logs = ...; // 日志数据流

DataStream<String> filteredLogs = logs.filter(new FilterFunction<String>() {
    @Override
    public boolean filter(String value) throws Exception {
        try {
            // 尝试解析为JSON格式
            JsonNode jsonNode = new ObjectMapper().readTree(value);
            return true;
        } catch (Exception e) {
            // 解析失败，说明是非JSON格式的字符串，过滤掉
            return false;
        }
    }
});

DataStream<Tuple2<Long, String>> timestampedLogs = filteredLogs.map(new MapFunction<String, Tuple2<Long, String>>() {
    @Override
    public Tuple2<Long, String> map(String value) throws Exception {
        JsonNode jsonNode = new ObjectMapper().readTree(value);
        long ts = jsonNode.get("ts").asLong();
        return Tuple2.of(ts, value);
    }
}).assignTimestampsAndWatermarks(new BoundedOutOfOrdernessTimestampExtractor<Tuple2<Long, String>>(Time.seconds(5)) {
    @Override
    public long extractTimestamp(Tuple2<Long, String> element) {
        return element.f0;
    }
});

在上面的代码中，首先使用filter算子过滤掉非JSON格式的字符串，然后使用map算子提取ts字段作为时间戳，并生成Tuple2<Long, String>类型的数据流。最后使用assignTimestampsAndWatermarks算子生成水位线，水位线的最大乱序程度为5秒