Python 过滤日志脏数据并提取最大时间戳生成水位线

为了过滤日志数据中的脏数据，需要编写一个程序来检查每个字符串是否为有效的 JSON 格式。可以使用 Python 的 json 模块来进行验证。如果字符串无法解析为 JSON 格式，则可以将它视为脏数据并忽略。

以下是可能的代码示例：

import json
import time

# 假设日志数据以字符串列表的形式存储在 logs 变量中
logs = [...]

# 初始化水位线和最大时间戳
watermark = 0
max_ts = 0

# 迭代日志数据并过滤脏数据
for log in logs:
    try:
        data = json.loads(log)
        ts = data.get('ts', 0)
        if ts > max_ts:
            max_ts = ts
    except ValueError:
        continue

# 计算水位线并添加最大乱序程度
watermark = max_ts - 5

在上面的代码中，我们首先使用 json.loads() 方法尝试将每个日志记录解析为 JSON 格式。如果解析成功，则我们提取其中的 'ts' 字段并将其与当前的最大时间戳进行比较。如果 'ts' 更大，则更新最大时间戳。如果解析失败，则我们将该记录视为脏数据并忽略它。

最后，我们使用最大时间戳减去 5 秒来计算水位线，并将其存储在 watermark 变量中。这个水位线将用于控制下游系统的数据处理速度，以确保不会处理过期的数据。