如何用Elasticsearch按Connection ID关联聚合Mongo日志生成用户-驱动列表？

针对你的需求，这里提供三种可行的实现方式，覆盖实时关联、持久化映射和临时可视化场景，同时处理Connection ID短时间重复的问题：

一、用Elasticsearch Transform构建持久化关联索引 #

Transform适合将分散的日志事件聚合为结构化的驱动-用户映射表，方便后续查询和可视化，还能自动同步新数据。

步骤： #

1. 创建Transform配置：按Connection ID分组，取每个ID对应的最新NETWORK（驱动版本）和ACCESS（用户名）事件，同时统计关联次数，加入时间过滤避免旧ID干扰：

PUT _transform/driver-user-association
{
  "source": {
    "index": ["mongo-logs"],
    "query": {
      "bool": {
        "filter": [
          {"range": {"@timestamp": {"gte": "now-7d"}}}, // 只处理最近7天的日志，可调整
          {"terms": {"event.type": ["NETWORK", "ACCESS"]}}
        ]
      }
    }
  },
  "pivot": {
    "group_by": {
      "conn_id": {"terms": {"field": "connection_id.keyword"}}
    },
    "aggregations": {
      "latest_driver": {
        "top_hits": {
          "size": 1,
          "sort": [{"@timestamp": "desc"}],
          "_source": {"includes": ["driver_version"]}
        }
      },
      "latest_user": {
        "top_hits": {
          "size": 1,
          "sort": [{"@timestamp": "desc"}],
          "_source": {"includes": ["username"]}
        }
      },
      "association_count": {"value_count": {"field": "@timestamp"}}
    }
  },
  "dest": {
    "index": "driver-user-mapping"
  },
  "sync": {
    "time": {
      "field": "@timestamp",
      "delay": "60s" // 每分钟同步一次新日志
    }
  }
}

2. 启动Transform：

POST _transform/driver-user-association/_start

3. 使用与导出：

• 在Kibana中创建driver-user-mapping的索引模式，用Data Table可视化展示latest_driver.hits.hits[0]._source.driver_version、latest_user.hits.hits[0]._source.username和association_count
• 点击可视化面板的「导出」按钮，选择CSV格式下载结果

二、Kibana可视化中直接聚合（临时场景） #

如果不需要持久化数据，可直接在Kibana中通过脚本字段+聚合实现关联，适合临时分析。

步骤： #

1. 添加脚本字段：
   进入Kibana「Stack Management」→「索引模式」→选择你的Mongo日志索引→「脚本字段」，添加两个字段：

   • 字段名：extracted_driver，脚本：

     if (doc['event.type'].value == 'NETWORK') {
       return doc['driver_version'].value;
     } else {
       return null;
     }
     

   • 字段名：extracted_user，脚本：

     if (doc['event.type'].value == 'ACCESS') {
       return doc['username'].value;
     } else {
       return null;
     }
     

2. 创建Data Table可视化：

   • 行分组：先选extracted_driver.keyword，再选extracted_user.keyword
   • 指标：选择「Unique Count」，字段选connection_id.keyword（统计每个驱动-用户对的唯一连接数），或选「Count」统计总事件数
   • 添加时间过滤器：比如只选最近24小时，避免Connection ID重复导致的错误关联

3. 导出CSV：在可视化面板右上角点击「导出」即可下载结果

三、用Enrich Policy实时关联日志（实时场景） #

如果需要在日志流入Elasticsearch时就完成关联，让每条ACCESS事件自动带上对应驱动版本，可使用Enrich Policy。

步骤： #

1. 创建Enrich源索引：存储NETWORK事件的Connection ID和驱动版本：

PUT /network-enrich-source
{
  "mappings": {
    "properties": {
      "connection_id": {"type": "keyword"},
      "driver_version": {"type": "keyword"},
      "@timestamp": {"type": "date"}
    }
  }
}

2. 导入NETWORK数据到源索引：

POST _reindex
{
  "source": {
    "index": "mongo-logs",
    "query": {"term": {"event.type": "NETWORK"}}
  },
  "dest": {
    "index": "network-enrich-source"
  }
}

3. 创建Enrich策略：只保留最近24小时的Connection ID，避免重复：

PUT _enrich/policy/driver-lookup-policy
{
  "match": {
    "indices": "network-enrich-source",
    "match_field": "connection_id",
    "enrich_fields": ["driver_version"],
    "query": {"range": {"@timestamp": {"gte": "now-24h"}}}
  }
}

4. 执行策略：

POST _enrich/policy/driver-lookup-policy/_execute

5. 创建Ingest Pipeline：将ACCESS事件与驱动版本关联：

PUT _ingest/pipeline/enrich-driver-pipeline
{
  "processors": [
    {
      "enrich": {
        "policy_name": "driver-lookup-policy",
        "field": "connection_id",
        "target_field": "enriched",
        "max_matches": 1 // 只取最近的匹配结果
      }
    },
    {
      "rename": {
        "field": "enriched.driver_version",
        "target_field": "driver_version",
        "ignore_missing": true
      }
    },
    {
      "remove": {
        "field": "enriched"
      }
    }
  ]
}

6. 应用Pipeline：

   • 新日志导入时，指定这个Pipeline；对现有日志，用_reindex重新处理：

     POST _reindex
     {
       "source": {
         "index": "mongo-logs",
         "query": {"term": {"event.type": "ACCESS"}}
       },
       "dest": {
         "index": "mongo-logs",
         "pipeline": "enrich-driver-pipeline"
       }
     }
     

7. 可视化与导出：此时ACCESS事件已包含driver_version，直接在Kibana中按driver_version.keyword和username.keyword聚合，生成列表并导出CSV。

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