Zeek连接数据加载到PyFlink时带点号id字段的处理问题

问题核心 #

Zeek输出的JSON数据为扁平结构，id类元组字段被序列化为带点号的键名（如id.orig_h），而Flink JSON解析器默认会将点号识别为嵌套结构的分隔符，直接解析会出现字段匹配失败、取值为空的问题。

可行处理方案 #

方案1：显式指定JSON字段路径（推荐，无需修改原始数据） #

建表时直接通过JSON路径表达式绑定带点号的键，可自定义列名避免点号后续对SQL语法的影响，示例代码如下：

CREATE TABLE zeek_conn (
    ts DOUBLE,
    uid STRING,
    -- 显式指定每个带点号字段对应的JSON路径
    id_orig_h STRING '$.id.orig_h',
    id_orig_p INT '$.id.orig_p',
    id_resp_h STRING '$.id.resp_h',
    id_resp_p INT '$.id.resp_p',
    proto STRING,
    conn_state STRING,
    missed_bytes INT,
    history STRING,
    orig_pkts INT,
    orig_ip_bytes INT,
    resp_pkts INT,
    resp_ip_bytes INT
) WITH (
    'connector' = 'filesystem',
    'path' = 'file:///path/to/zeek/json/files',
    'format' = 'json',
    'json.ignore-parse-errors' = 'true' -- 可选，容忍脏数据
);

如果使用PyFlink DataStream API，可以通过JsonRowDeserializationSchema的jsonSchema配置指定字段映射，示例：

from pyflink.common.typeinfo import Types
from pyflink.datastream.connectors.file_system import FileSource, StreamFormat
from pyflink.common.serialization import JsonRowDeserializationSchema

deserialization_schema = JsonRowDeserializationSchema.builder() \
    .type_info(Types.ROW_NAMED(
        ["ts", "uid", "id_orig_h", "id_orig_p", "id_resp_h", "id_resp_p", "proto"],
        [Types.DOUBLE(), Types.STRING(), Types.STRING(), Types.INT(), Types.STRING(), Types.INT(), Types.STRING()]
    )) \
    .json_schema("""
    {
        "type": "object",
        "properties": {
            "ts": {"type": "number"},
            "uid": {"type": "string"},
            "id.orig_h": {"type": "string"},
            "id.orig_p": {"type": "integer"},
            "id.resp_h": {"type": "string"},
            "id.resp_p": {"type": "integer"},
            "proto": {"type": "string"}
        }
    }
    """) \
    .build()

方案2：预处理JSON键名替换点号 #

如果带点号的字段数量多，可以先写一个简单的UDF对原始JSON字符串做预处理，统一将键中的点号替换为下划线，再进行后续解析：

from pyflink.common.functions import MapFunction
import json

class ReplaceDotKeyMapFunction(MapFunction):
    def map(self, value):
        json_data = json.loads(value)
        new_json = {}
        for k, v in json_data.items():
            new_k = k.replace('.', '_')
            new_json[new_k] = v
        return json.dumps(new_json)

# 调用示例，先对原始JSON流做转换再解析
ds = env.read_text_file('/path/to/zeek.json')
converted_ds = ds.map(ReplaceDotKeyMapFunction(), output_type=Types.STRING())
# 后续直接用下划线键名解析即可

方案3：映射为ROW类型对齐Zeek原生结构 #

如果需要保留Zeek原生的元组结构语义，可以先通过JSON路径将分散的扁平字段组装为ROW类型字段：

CREATE TABLE zeek_conn (
    ts DOUBLE,
    uid STRING,
    id ROW<orig_h STRING, orig_p INT, resp_h STRING, resp_p INT>,
    proto STRING,
    -- 其余字段省略
) WITH (
    'connector' = 'filesystem',
    'path' = 'file:///path/to/zeek/json/files',
    'format' = 'json',
    -- 显式绑定ROW子字段对应的JSON路径
    'json.fields' = 'ts,uid,id.orig_h,orig_p:id.orig_p,id.resp_h,resp_p:id.resp_p,proto'
);

内容的提问来源于stack exchange，提问作者Ben