« ES|QL limitations SQL »
Elastic Docs ›Elasticsearch Guide ›ES|QL

ES|QL 示例

edit

ES|QL 示例

edit

聚合和丰富Windows事件日志

edit
FROM logs-*
| WHERE event.code IS NOT NULL
| STATS event_code_count = COUNT(event.code) BY event.code,host.name
| ENRICH win_events ON event.code WITH event_description
| WHERE event_description IS NOT NULL and host.name IS NOT NULL
| RENAME event_description AS event.description
| SORT event_code_count DESC
| KEEP event_code_count,event.code,host.name,event.description
  • 它首先从匹配模式 "logs-*" 的索引中查询日志。
  • 过滤 "event.code" 字段不为空的事件。
  • 按 "event.code" 和 "host.name" 聚合事件的计数。
  • 使用 "EVENT_DESCRIPTION" 字段为事件添加额外信息。
  • 过滤掉 "EVENT_DESCRIPTION" 或 "host.name" 为空的事件。
  • 将 "EVENT_DESCRIPTION" 重命名为 "event.description"。
  • 按 "event_code_count" 降序排序结果。
  • 仅保留选定的字段："event_code_count"、"event.code"、"host.name" 和 "event.description"。

汇总来自进程 curl.exe 的出站流量

edit
FROM logs-endpoint
| WHERE process.name == "curl.exe"
| STATS bytes = SUM(destination.bytes) BY destination.address
| EVAL kb =  bytes/1024
| SORT kb DESC
| LIMIT 10
| KEEP kb,destination.address
  • 从“logs-endpoint”源查询日志。
  • 过滤“process.name”字段为“curl.exe”的事件。
  • 计算发送到目标地址的字节总和并将其转换为千字节（KB）。
  • 按“kb”（千字节）降序排列结果。
  • 将输出限制为前10个结果。
  • 仅保留“kb”和“destination.address”字段。

操作DNS日志以查找每个注册域名的高数量唯一DNS查询

edit
FROM logs-*
| GROK dns.question.name "%{DATA}\\.%{GREEDYDATA:dns.question.registered_domain:string}"
| STATS unique_queries = COUNT_DISTINCT(dns.question.name) BY dns.question.registered_domain, process.name
| WHERE unique_queries > 10
| SORT unique_queries DESC
| RENAME unique_queries AS `Unique Queries`, dns.question.registered_domain AS `Registered Domain`, process.name AS `Process`
  • 从匹配“logs-*”的索引中查询日志。
  • 使用“grok”模式从“dns.question.name”字段中提取注册域名。
  • 计算每个注册域名和进程名称的唯一DNS查询计数。
  • 过滤“unique_queries”大于10的结果。
  • 按“unique_queries”降序排列结果。
  • 为清晰起见重命名字段：“unique_queries”为“Unique Queries”，“dns.question.registered_domain”为“Registered Domain”，“process.name”为“Process”。

识别大量出站用户连接

edit
FROM logs-*
| WHERE NOT CIDR_MATCH(destination.ip, "10.0.0.0/8", "172.16.0.0/12", "192.168.0.0/16")
| STATS destcount = COUNT(destination.ip) BY user.name, host.name
| ENRICH ldap_lookup_new ON user.name
| WHERE group.name IS NOT NULL
| EVAL follow_up = CASE(destcount >= 100, "true","false")
| SORT destcount DESC
| KEEP destcount, host.name, user.name, group.name, follow_up
  • 从匹配“logs-*”的索引中查询日志。
  • 过滤掉目标IP地址在私有IP地址范围内的日志（例如，10.0.0.0/8、172.16.0.0/12、192.168.0.0/16）。
  • 按“user.name”和“host.name”计算唯一目标IP的数量。
  • 使用LDAP组信息丰富“user.name”字段。
  • 过滤掉“group.name”为空的结果。
  • 使用“CASE”语句创建一个“follow_up”字段，当“destcount”大于或等于100时设置为“true”，否则设置为“false”。
  • 按“destcount”降序排列结果。
  • 保留选定的字段：“destcount”、“host.name”、“user.name”、“group.name”和“follow_up”。
« ES|QL limitations SQL »