如何在Kibana中可视化多值字段的端口计数统计？

当然可以实现这种跨多字段的端口计数可视化，下面分两种方案给你讲解：直接在Kibana中配置的快速方案，以及更高效的数据重构方案。

方案一：用Kibana Runtime Fields快速实现（无需修改原始数据） #

Runtime Fields可以在不重新索引数据的前提下，动态生成包含所有端口的数组字段，完美适配你的场景：

1. 进入Kibana的Stack Management > Index Patterns，找到你的目标索引模式。
2. 点击Add field > Runtime field，字段类型选择Long（端口是数字类型，统计更准确），给字段命名比如all_ports。
3. 在脚本编辑器中写入Painless脚本，把所有相关端口字段的内容合并成一个数组——分两种情况：

   情况1：已将ip1Ports/ip2Ports拆分成column1/column2等字段 #

   def ports = [];
   // 收集ip1的所有端口
   if (doc['ip1Ports_List.column1'].size() > 0) ports.add(doc['ip1Ports_List.column1'].value);
   if (doc['ip1Ports_List.column2'].size() > 0) ports.add(doc['ip1Ports_List.column2'].value);
   if (doc['ip1Ports_List.column3'].size() > 0) ports.add(doc['ip1Ports_List.column3'].value);
   // 收集ip2的所有端口
   if (doc['ip2Ports_List.column1'].size() > 0) ports.add(doc['ip2Ports_List.column1'].value);
   if (doc['ip2Ports_List.column2'].size() > 0) ports.add(doc['ip2Ports_List.column2'].value);
   // 过滤掉不需要统计的0端口（如果需要保留可以删掉这行）
   return ports.stream().filter(p -> p != 0).collect(Collectors.toList());
   

   情况2：仍是原始的逗号分隔字符串字段（比如ip1Ports = "80,57000,0"） #

   def ports = [];
   // 处理ip1Ports字段
   if (doc['ip1Ports'].size() > 0) {
       def ip1Parts = doc['ip1Ports'].value.splitOnToken(',');
       for (part in ip1Parts) {
           def port = Integer.parseInt(part.trim());
           if (port != 0) ports.add(port);
       }
   }
   // 处理ip2Ports字段
   if (doc['ip2Ports'].size() > 0) {
       def ip2Parts = doc['ip2Ports'].value.splitOnToken(',');
       for (part in ip2Parts) {
           def port = Integer.parseInt(part.trim());
           if (port != 0) ports.add(port);
       }
   }
   return ports;
   

4. 保存这个Runtime字段，回到可视化界面创建新的可视化（比如垂直条形图）：
   • 选择Metrics为Count
   • 选择Buckets > X-Axis为Terms聚合，字段选择刚创建的all_ports
   • 调整排序和数量限制后，就能看到所有端口的总出现次数统计了

方案二：重构数据（长期更高效） #

如果你的数据量较大，Runtime Fields的动态计算可能会影响查询性能，这时候建议在数据摄入阶段就把端口处理成数组字段，比如用Elasticsearch的Ingest Pipeline：

1. 进入Kibana的Stack Management > Ingest Pipelines，创建新管道。
2. 添加以下处理器（可直接复制JSON）：

   {
     "description": "Merge port fields into a single array",
     "processors": [
       {
         "split": {
           "field": "ip1Ports",
           "separator": ",",
           "target_field": "ip1Ports_array",
           "ignore_missing": true
         }
       },
       {
         "split": {
           "field": "ip2Ports",
           "separator": ",",
           "target_field": "ip2Ports_array",
           "ignore_missing": true
         }
       },
       {
         "script": {
           "source": """
             def allPorts = [];
             if (ctx.ip1Ports_array != null) allPorts.addAll(ctx.ip1Ports_array);
             if (ctx.ip2Ports_array != null) allPorts.addAll(ctx.ip2Ports_array);
             // 转换为整数并过滤0端口
             ctx.all_ports = allPorts.stream().map(p -> Integer.parseInt(p.trim())).filter(p -> p != 0).collect(Collectors.toList());
             // 删除临时字段（可选）
             ctx.remove('ip1Ports_array');
             ctx.remove('ip2Ports_array');
           """
         }
       }
     ]
   }
   

3. 保存管道后，将其应用到你的数据摄入流程（比如在Beats或Logstash中指定这个管道，或者对已有数据重新索引时使用）。
4. 之后创建可视化时，直接对all_ports字段做Terms聚合即可，性能比Runtime Fields更好。

为什么你之前的方法行不通？ #

• 拆分后的column1/column2是独立字段，Terms聚合只能针对单个字段统计，无法自动合并多个字段的数值；
• 拆分图表是按字段维度拆分统计，得到的是每个字段下端口的计数，而不是所有端口的总计数；
• 原始字符串字段的聚合是按完整字符串（比如"80,443"）作为一个项统计，不会拆分内部的端口值。

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