日志服务数据加工最佳实践: 多子键为数组的复杂JSON加工

本文涉及的产品
对象存储 OSS,20GB 3个月
文件存储 NAS,50GB 3个月
对象存储 OSS,内容安全 1000 次 1年
简介: 程序构建的日志经常会以一种统计性质的JSON格式写入, 通常其包含一个基础信息, 以及多个子健为数组的形式. 本篇如何使用日志服务数据加工处理多子键为数组的复杂JSON.

加工需求

统计类日志形式

程序构建的日志经常会以一种统计性质的JSON格式写入, 通常其包含一个基础信息, 以及多个子健为数组的形式. 例如一个服务器每隔1分钟写入一条日志, 包含当前信息状态, 以及相关服务器和客户端节点的统计状态信息.

样例:

__source__:  1.2.3.4
__topic__:  
content:{
     "service": "search_service",
     "overal_status": "yellow",
     "servers": [
         {
             "host": "1.2.3.4",
             "status": "green"
         },
         {
             "host": "1.2.3.5",
             "status": "green"
         }
     ],
     "clients": [
         {
             "host": "1.2.3.6",
             "status": "green"
         },
         {
             "host": "1.2.3.7",
             "status": "red"
         }
     ]
}

加工需求

1、对原始日志进行topic分裂,主题主要分为三个,分别是overall_type、client_status、server_status
2、对于不同的topic保存不同的信息

  • overall_type:保留server、client数量,overal_status颜色和service信息
  • client_status: 保留对应的host地址、status状态和service信息
  • server_status: 保留对应的host地址、status状态和service信息

期望输出的日志

期望样例中的一条日志会被分裂加工成5条日志:

__source__:  1.2.3.4
__topic__:  overall_type
client_count:  2
overal_status:  yellow
server_count:  2
service:  search_service


__source__:  1.2.3.4
__topic__:  client_status
host:  1.2.3.7
status:  red
service:  search_service


__source__:  1.2.3.4
__topic__:  client_status
host:  1.2.3.6
status:  green
service:  search_service


__source__:  1.2.3.4
__topic__:  server_status
host:  1.2.3.4
status:  green
service:  search_service


__source__:  1.2.3.4
__topic__:  server_status
host:  1.2.3.5
status:  green
service:  search_service

解决方案

初步处理

1、第一步将一条日志拆分成3条日志, 给主题赋予3个不同值, 在进行分裂,经过分裂后会分成除了topic不同,其他信息相同的三条日志。

e_set("__topic__", "server_status,client_status,overall_type")
e_split("__topic__")

处理后日志格式如下(在内存中):

__source__:  1.2.3.4
__topic__:  server_status                    // 另外2条是client_status和overall_type, 其他一样
content:  {
    ...如前...
}

2、第二步为基于content的JSON内容在第一层展开, 并删除content字段:

e_json('content',depth=1)
e_drop_fields("content")

处理后的日志格式如下(在内存中):

__source__:  1.2.3.4
__topic__:  overall_type                          // 另外2条是client_status和overall_type, 其他一样
clients:  [{"host": "1.2.3.6", "status": "green"}, {"host": "1.2.3.7", "status": "red"}]
overal_status:  yellow
servers:  [{"host": "1.2.3.4", "status": "green"}, {"host": "1.2.3.5", "status": "green"}]
service:  search_service  

处理overall_type日志

  1. 针对主题是overall_type的日志, 统计client_count和server_count:
e_if(e_search("__topic__==overall_type"), 
       e_compose(
                 e_set("client_count" json_select(v("clients"), "length([*])", default=0)), 
                 e_set("server_count" json_select(v("servers"), "length([*])", default=0))
    ))

处理后的日志为(仅显示修改部分):

__topic__:  overall_type
server_count:  2
client_count:  2
  1. 丢弃相关字段:
e_if(e_search("__topic__==overall_type"), e_drop_fields("clients", "servers"))

处理server_status日志

  1. 针对主题是server_status的日志, 进行进一步分裂.
e_if(e_search("__topic__==server_status"), 
       e_compose(
                 e_split("servers"), 
                 e_json("servers", depth=1)
    ))

处理后的日志为2条如下(仅显示修改部分):

__topic__:  server_status
servers:  {"host": "1.2.3.4", "status": "green"}
host: 1.2.3.4
status: green

__topic__:  server_status
servers:  {"host": "1.2.3.5", "status": "green"}
host: 1.2.3.5
status: green
  1. 保留相关字段:
e_if(e_search("__topic__==overall_type"), e_drop_fields("servers"))

处理client_status日志

  1. 同理, 针对主题是client_status的日志, 进行进一步分裂, 在删除多余字段.
e_if(e_search("__topic__==client_status"), 
       e_compose(
                 e_split("clients"), 
                 e_json("clients", depth=1),
                 e_drop_fields("clients")
    ))

处理后的日志为2条如下(仅显示修改部分):

__topic__:  client_status
host: 1.2.3.6
status: green

__topic__:  clients
host: 1.2.3.7
status: red

综合

综上,LOG DSL规则是


# 总体分裂
e_set("__topic__", "server_status,client_status,overall_type")
e_split("__topic__")
e_json('content',depth=1)
e_drop_fields("content")

# 处理overall_type日志
e_if(e_search("__topic__==overall_type"), 
       e_compose(
                 e_set("client_count" json_select(v("clients"), "length([*])", default=0)), 
                 e_set("server_count" json_select(v("servers"), "length([*])", default=0))
    ))

# 处理server_status日志
e_if(e_search("__topic__==server_status"), 
       e_compose(
                 e_split("servers"), 
                 e_json("servers", depth=1)
    ))
e_if(e_search("__topic__==overall_type"), e_drop_fields("servers"))


# 处理client_status日志
e_if(e_search("__topic__==client_status"), 
       e_compose(
                 e_split("clients"), 
                 e_json("clients", depth=1),
                 e_drop_fields("clients")
    ))

方案优化

一个边界问题

注意到以上方案对于content.serverscontent.servers是空时的处理有一些问题,

假设原始日志是:

__source__:  1.2.3.4
__topic__:  
content:{
            "service": "search_service",
            "overal_status": "yellow",
            "servers": [ ],
            "clients": [ ]
}

会被分裂为3条日志, 其中主题是client_status和server_status的日志内容是空的.

__source__:  1.2.3.4
__topic__:  overall_type
client_count:  0
overal_status:  yellow
server_count:  0
service:  search_service


__source__:  1.2.3.4
__topic__:  client_status
service:  search_service
__source__:  1.2.3.4


__topic__:  server_status
host:  1.2.3.4
status:  green
service:  search_service

方案1

这里可以在初始分裂后, 处理server_statusclient_status日志前分别判断并丢弃空的相关事件:

# 处理server_status: 空的丢弃(非空保留)
e_keep(op_and(e_search("__topic__==server_status"), json_select(v("servers"), "length([*])")))

# 处理client_status: 空的丢弃(非空保留)
e_keep(op_and(e_search("__topic__==client_status"), json_select(v("clients"), "length([*])")))

综合

综上,LOG DSL规则是


# 总体分裂
e_set("__topic__", "server_status,client_status,overall_type")
e_split("__topic__")
e_json('content',depth=1)
e_drop_fields("content")

# 处理overall_type日志
e_if(e_search("__topic__==overall_type"), 
       e_compose(
                 e_set("client_count" json_select(v("clients"), "length([*])", default=0)), 
                 e_set("server_count" json_select(v("servers"), "length([*])", default=0))
    ))

# 新加: 预处理server_status: 空的丢弃(非空保留) 
e_keep(op_and(e_search("__topic__==server_status"), json_select(v("servers"), "length([*])")))

# 处理server_status日志
e_if(e_search("__topic__==server_status"), 
       e_compose(
                 e_split("servers"), 
                 e_json("servers", depth=1)
    ))
e_if(e_search("__topic__==overall_type"), e_drop_fields("servers"))


# 新加: 预处理client_status: 空的丢弃(非空保留) 
e_keep(op_and(e_search("__topic__==client_status"), json_select(v("clients"), "length([*])")))

# 处理client_status日志
e_if(e_search("__topic__==client_status"), 
       e_compose(
                 e_split("clients"), 
                 e_json("clients", depth=1),
                 e_drop_fields("clients")
    ))

方案2

在初始分裂时进行判断, 如果对应数据是空的就不分裂出更多事件:

# 初始主题
e_set("__topic__", "server_status")

# 如果content.servers非空, 则从server_status分裂出1条日志
e_if(json_select(v("content"), "length(servers[*])"),
     e_compse(
           e_set("__topic__", "server_status,overall_type"),
           e_split("__topic__")
     ))

# 如果content.clients非空, 则从overall_type再分裂出1条日志
e_if(op_and(e_search("__topic__==overall_type"), json_select(v("content"), "length(clients[*])")),
     e_compse(
           e_set("__topic__", "client_status,overall_type"),
           e_split("__topic__")
     ))

综合

综上,LOG DSL规则是


# 总体分裂
e_set("__topic__", "server_status")

# 如果content.servers非空, 则从server_status分裂出1条日志
e_if(json_select(v("content"), "length(servers[*])"),
     e_compse(
           e_set("__topic__", "server_status,overall_type"),
        e_split("__topic__")
     ))

# 如果content.clients非空, 则从server_status分裂出1条日志
e_if(op_and(e_search("__topic__==overall_type"), json_select(v("content"), "length(clients[*])")),
     e_compse(
           e_set("__topic__", "client_status,overall_type"),
        e_split("__topic__")
     ))

# 处理overall_type日志
e_if(e_search("__topic__==overall_type"), 
       e_compose(
                 e_set("client_count" json_select(v("clients"), "length([*])", default=0)), 
                 e_set("server_count" json_select(v("servers"), "length([*])", default=0))
    ))

# 处理server_status日志
e_if(e_search("__topic__==server_status"), 
       e_compose(
                 e_split("servers"), 
                 e_json("servers", depth=1)
    ))
e_if(e_search("__topic__==overall_type"), e_drop_fields("servers"))


# 处理client_status日志
e_if(e_search("__topic__==client_status"), 
       e_compose(
                 e_split("clients"), 
                 e_json("clients", depth=1),
                 e_drop_fields("clients")
    ))

比较

方案1会在分裂出日志后再删除, 逻辑上有些多余, 但规则简单易维护. 默认推荐.
方案2会在分裂前进行判断, 处理效率会高一些, 但规则略微冗余, 仅在特定场景(例如初始分裂可能导致大量额外事件产生)时推荐.

进一步参考

欢迎扫码加入官方钉钉群获得实时更新与阿里云工程师的及时直接的支持:
image

相关实践学习
【涂鸦即艺术】基于云应用开发平台CAP部署AI实时生图绘板
【涂鸦即艺术】基于云应用开发平台CAP部署AI实时生图绘板
目录
相关文章
|
JavaScript 测试技术 API
从创建到维护:掌握package.json的最佳实践(二)
从创建到维护:掌握package.json的最佳实践
|
存储 JSON JavaScript
从创建到维护:掌握package.json的最佳实践(一)
从创建到维护:掌握package.json的最佳实践
|
10天前
|
JSON 安全 API
Python处理JSON数据的最佳实践:从基础到进阶的实用指南
JSON作为数据交换通用格式,广泛应用于Web开发与API交互。本文详解Python处理JSON的10个关键实践,涵盖序列化、复杂结构处理、性能优化与安全编程,助开发者高效应对各类JSON数据挑战。
63 1
|
SQL 运维 监控
SLS 数据加工全面升级,集成 SPL 语法
在系统开发、运维过程中,日志是最重要的信息之一,其最大的优点是简单直接。SLS 数据加工功能旨在解决非结构化的日志数据处理,当前全面升级,集成 SPL 语言、更强的数据处理性能、更优的使用成本。
18422 224
|
JSON JavaScript 数据格式
1.js动态的往json数据添加新属性和值 2.JSON 和 JS 对象互转 3.对象转化为数组
1.js动态的往json数据添加新属性和值 2.JSON 和 JS 对象互转 3.对象转化为数组
113 0
|
SQL JSON 监控
实时计算 Flink版产品使用合集之直接将 JSON 字符串解析为数组的内置函数如何解决
实时计算Flink版作为一种强大的流处理和批处理统一的计算框架,广泛应用于各种需要实时数据处理和分析的场景。实时计算Flink版通常结合SQL接口、DataStream API、以及与上下游数据源和存储系统的丰富连接器,提供了一套全面的解决方案,以应对各种实时计算需求。其低延迟、高吞吐、容错性强的特点,使其成为众多企业和组织实时数据处理首选的技术平台。以下是实时计算Flink版的一些典型使用合集。
|
JSON JavaScript Go
JSON文件读写最佳实践
JSON文件读写最佳实践
124 0
|
JSON 数据格式
ECharts折线图渲染json格式数据(格式为数组)
ECharts折线图渲染json格式数据(格式为数组)
174 0
|
JSON 编解码 JavaScript
工银e生活开发脱坑日志(8)使用json_decode无法解析json,双引号才是json 的标准
工银e生活开发脱坑日志(8)使用json_decode无法解析json,双引号才是json 的标准
209 0
|
消息中间件 JSON 缓存
Golang:将日志以Json格式输出到Kafka2
Golang:将日志以Json格式输出到Kafka2
286 0

相关产品

  • 日志服务