如何用jq在连续JSON记录流中按指定条数触发动作并实时输出

核心思路 #

利用jq的流式遍历与状态管理，在每条JSON记录实时输出的同时，累计指定数量的记录后自动统计非null记录数，完全适配每秒输出一条的场景。

bash函数实现 #

alert() {
    local batch_size="$1"
    # 校验输入参数是否为正整数
    if ! [[ "$batch_size" =~ ^[1-9][0-9]*$ ]]; then
        echo "错误：batch_size必须是正整数" >&2
        return 1
    fi

    jq --unbuffered --argjson n "$batch_size" '
        begin
            $n as $batch |
            # 初始化状态：[当前批次计数, 当前批次记录数组]
            [0, []] as $state
        end
        | foreach . as $item ($state;
            # 更新状态：计数+1，当前记录加入批次数组
            [.[0] + 1, .[1] + [$item]],
            # 输出原始JSON记录，满足每秒输出需求
            $item,
            # 判断是否达到批次阈值，触发统计
            if .[0] + 1 == $batch then
                # 统计当前批次中非null的记录数
                (.[1] + [$item] | map(select(. != null)) | length) as $valid_count
                | "=== 累计\($batch)条记录，非null记录数：\($valid_count) ==="
                | [0, []]  # 重置状态，准备下一批次
            else
                empty  # 未达阈值，无额外输出
            end
        )
        # 处理最后一批不足batch_size的剩余记录
        | end
        | if .[1] | length > 0 then
            (.[1] | map(select(. != null)) | length) as $valid_count
            | "=== 剩余\(.[1] | length)条记录，非null记录数：\($valid_count) ==="
          else
            empty
          end
    '
}

使用方法 #

将你的重放管道与alert函数结合，指定批次大小（比如x=5）：

( read line ; while [[ -n "${line}" ]] ; do echo "${line}"; sleep 1 ; read line ; done ) < json.data | alert 5

关键细节说明 #

1. --unbuffered参数：强制jq实时处理每条输入，避免因缓存导致输出延迟，完美匹配每秒一条的输出节奏。
2. 状态管理：通过foreach维护批次计数和记录数组，确保每批次的记录被正确累计。
3. 非null判断：用jq原生的select(. != null)准确识别JSON null值，避免bash字符串处理的误判（比如字符串"null"和JSON null的区别）。
4. 剩余记录处理：在输入结束时自动统计最后一批不足指定数量的记录，避免数据遗漏。

x=5时的输出示例 #

假设json.data内容为：

{"id":1}
null
{"id":2}
{"id":3}
null
{"id":4}
{"id":5}
{"id":6}
null
{"id":7}

执行命令后的输出为：

{"id":1}
null
{"id":2}
{"id":3}
null
=== 累计5条记录，非null记录数：3 ===
{"id":4}
{"id":5}
{"id":6}
null
{"id":7}
=== 剩余5条记录，非null记录数：4 ===

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