如何将含LEFT JOIN的SQL查询转换为Elastic DSL语句

先把冗余逻辑筛除，原SQL两处cnt >= 1的过滤是无效的：按SrcIp、DstIp字段做GROUP BY后，每个分组的计数值最小为1，加该过滤不会改变返回结果。
核心要实现的逻辑等价于：

• 从所有SubLogClass = "syslog"的日志中，筛选两类IP访问对聚合结果：
  • 时间窗口0：LogTime >= "11111111"，按(SrcIp, DstIp)聚合，统计每组日志数cnt、最早访问时间min_time
  • 时间窗口1：LogTime >= "22222222"，按(SrcIp, DstIp)聚合，统计每组日志数cnt、最早访问时间min_time
• 做关联匹配：当窗口0的DstIp等于窗口1的SrcIp，且窗口0记录的min_time <= 窗口1记录的min_time时，返回所有匹配成功的窗口0聚合记录。

原SQL写的LEFT JOIN加tt_1.SrcIp IS NOT NULL本质就是内连接效果，没有特殊逻辑。

ES作为分布式检索引擎，不支持关系型数据库任意维度的跨结果集JOIN，不要照搬SQL的子查询写法，根据你用的ES版本选对应方案即可：

方案1：兼容ES 7.x及以上所有版本 #

用filters聚合+composite聚合，在一次查询里完成两个时间窗口的IP对聚合计算，因为聚合是在各分片本地执行，最终返回的聚合结果数据量远小于原始日志量，性能损耗极低。
查询DSL如下：

POST /你的日志索引名/_search
{
  "size": 0,
  "query": {
    "bool": {
      "filter": [
        {"term": {"SubLogClass": "syslog"}},
        {"range": {"LogTime": {"gte": "11111111"}}}
      ]
    }
  },
  "aggs": {
    "split_time_window": {
      "filters": {
        "filters": {
          "window_0": {"range": {"LogTime": {"gte": "11111111"}}},
          "window_1": {"range": {"LogTime": {"gte": "22222222"}}}
        }
      },
      "aggs": {
        "ip_pair_group": {
          "composite": {
            "size": 10000,
            "sources": [
              {"SrcIp": {"terms": {"field": "SrcIp.keyword"}}},
              {"DstIp": {"terms": {"field": "DstIp.keyword"}}}
            ]
          },
          "aggs": {
            "cnt": {"value_count": {"field": "time"}},
            "min_time": {"min": {"field": "time"}}
          }
        }
      }
    }
  }
}

返回结果后的匹配逻辑非常简单：

• 把window_0下的所有IP对聚合结果存入临时字典，字典key为DstIp，value为对应记录的SrcIp、cnt、min_time列表
• 遍历window_1下的所有IP对聚合记录，拿每条记录的SrcIp去临时字典匹配，只要存在min_time小于等于当前记录min_time的条目，就把该条目加入最终结果集，去重后就是和原SQL完全一致的返回。
  如果IP对总数量超过composite单页10000的上限，只要带上返回的after_key参数续查即可，全程不需要扫原始日志。

方案2：ES 8.4及以上版本直接用ES|QL实现 #

8.4版本后官方推出的ES|QL原生支持类SQL的JOIN逻辑，可以单条语句直接返回最终结果，不需要手动做结果匹配，还能裁剪原SQL里的冗余聚合逻辑，性能更好：

FROM 你的日志索引名
| WHERE SubLogClass == "syslog" AND LogTime >= "11111111"
| STATS cnt = count(*), min_time = min(time) BY SrcIp, DstIp
| JOIN (
    FROM 你的日志索引名
    | WHERE SubLogClass == "syslog" AND LogTime >= "22222222"
    | STATS min_time_t1 = min(time) BY SrcIp
  ) ON DstIp == SrcIp
| WHERE min_time <= min_time_t1
| KEEP SrcIp, DstIp, cnt, min_time

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