使用 jq 合并 json 对象流

Question

前提：希望从 json 日志文件中解析对象流，并根据特定条件输出“id.orig_h”连接到“id.resp_h”的总次数并显示总数。

示例 json 输入：

jq --slurp --raw-output . 
   
  {
    "ts": 1636606.998991,
    "uid": "CgbTrLvhqHAa",
    "id.orig_h": "10.8.21.11",
    "id.orig_p": 54858,
    "id.resp_h": "10.8.21.66",
    "id.resp_p": 5044,
    "proto": "tcp",
    "conn_state": "S0",
    "local_orig": true,
    "local_resp": true,
    "missed_bytes": 0,
    "history": "S",
    "orig_pkts": 1,
    "orig_ip_bytes": 60,
    "resp_pkts": 0,
    "resp_ip_bytes": 0
  },
  {
    "ts": 1636638.028568,
    "uid": "CFNumGx3XYWW7",
    "id.orig_h": "fe80::ba:61:fe3f:80",
    "id.orig_p": 130,
    "id.resp_h": "ff02::1",
    "id.resp_p": 131,
    "proto": "icmp",
    "duration": 3420.447889374,
    "orig_bytes": 2608,
    "resp_bytes": 0,
    "conn_state": "OTH",
    "local_orig": false,
    "local_resp": false,
    "missed_bytes": 0,
    "orig_pkts": 163,
    "orig_ip_bytes": 11736,
    "resp_pkts": 0,
    "resp_ip_bytes": 0
  },
  {
    "ts": 1636526872.598889,
    "uid": "Cq9JTE1OweOW6mi",
    "id.orig_h": "fe::63:88:14f5:b5",
    "id.orig_p": 131,
    "id.resp_h": "ff02::fb",
    "id.resp_p": 130,
    "proto": "icmp",
    "duration": 81086.88094513,
    "orig_bytes": 64000,
    "resp_bytes": 0,
    "conn_state": "OTH",
    "local_orig": false,
    "local_resp": false,
    "missed_bytes": 0,
    "orig_pkts": 4000,
    "orig_ip_bytes": 288000,
    "resp_pkts": 0,
    "resp_ip_bytes": 0
  },
  {
    "ts": 1636604547.798971,
    "uid": "Cs41IjaZTAdF7f",
    "id.orig_h": "fe::63:88:14f5:b5",
    "id.orig_p": 131,
    "id.resp_h": "ff02::1:ff:b5",
    "id.resp_p": 130,
    "proto": "icmp",
    "duration": 3414.3990546265,
    "orig_bytes": 2608,
    "resp_bytes": 0,
    "conn_state": "OTH",
    "local_orig": false,
    "local_resp": false,
    "missed_bytes": 0,
    "orig_pkts": 163,
    "orig_ip_bytes": 11736,
    "resp_pkts": 0,
    "resp_ip_bytes": 0
   }

我相信条件部分是好的

    jq -r '. | select(.resp_ip_bytes > 0 and .orig_ip_bytes > 0 and .duration > 0 and .orig_bytes > 0 and .resp_bytes >0)'

但是每次我尝试一个

    group_by([."id.orig_h", ."id.resp_h"]), 

getting --> 不能用字符串“id.orig_h”索引数字

期望的输出：

1.1.1.1 -> 2.2.2.2 | XXXX <- # of times

这里是没有join(" ")的输出

jq -sr 'map(select(.resp_ip_bytes > 0 and .orig_ip_bytes > 0 and .duration > 0 and .orig_bytes > 0 and .resp_bytes >0)) | group_by([."id.orig_h", ."id.resp_h"]) | map(length as $count | .[] | .count = $count) | sort_by([-.count, -.resp_ip_bytes]) | first | [."id.orig_h", "->", ."id.resp_h", "|", .count]'
[
  "10.8.21.11",
  "->",
  "10.8.21.123",
  "|",
  225 <--(not sure it matters but output on .count is yellow, all other output is green)
]

用join(" ")

string (" ") and number (225) cannot be added

Answer 1

试试这个

jq --slurp --raw-output '
  map(select(.resp_ip_bytes > 0 ... your conditions here ...))
  | group_by([."id.orig_h", ."id.resp_h"])
  | map(length as $count | .[] | .count = $count)
  | sort_by([-.count, -.resp_ip_bytes]) | first
  | [."id.orig_h", "->", ."id.resp_h", "|", .count]
  | join(" ")
'

这会插入另一个字段值count 与连接数，然后首先按最高的count 排序，然后按最高的resp_ip_bytes 排序，取第一个匹配项并根据需要格式化输出。

Answer 2

虽然内置的group_by 很方便，但它在空间和时间方面都非常低效，下面的替代方案可能特别适合非常长的日志文件。

注意inputs 与 -n 命令行选项一起使用：

< log.json jq -nr '
# Emit a stream of arrays, each array being a group defined by a value of f,
# which can be any jq filter that produces exactly one value for each item in `stream`.
def GROUPS_BY(stream; f): 
   reduce stream as $x ({};
     ($x|f) as $s
     | ($s|type) as $t
     | (if $t == "string" then $s else ($s|tojson) end) as $y
     | .[$t][$y] += [$x] )
   | .[][] ;

GROUPS_BY(inputs
          | select(.resp_ip_bytes > 0 and
                   .orig_ip_bytes > 0 and
                   .duration > 0 and .orig_bytes > 0 and
                   .resp_bytes >0);
          [."id.orig_h", ."id.resp_h"] ) 
| (first | "\(."id.orig_h") -> \(."id.resp_h")" ) +
  ( .[]  | " | \(length)" )
'

---

注意事项：

a) 上面定义的GROUPS_BY 是面向流的，无论是就其输入和输出而言。除此之外，GROUPS_BY 和group_by 的主要功能区别在于后者涉及排序。

b) 上面定义的GROUPS_BY/2 相对复杂，因为它被设计为具有group_by 的全部通用性，几乎可以作为插件替代。具体来说，E | group_by(F) 在功能上等同于：

[GROUPS_BY(E[]; F)] | sort_by(F)