Elasticsearch按时间查指定文档：求每个filename首尾文档优化方案

你的场景是4亿条数据量下用Spark窗口函数取每个filename的最早/最晚文档，这种全量扫描+窗口排序的方式确实会因为shuffle和大量数据传输导致性能极差。结合你使用的Elasticsearch 6.0.0和Spark 2.2技术栈，我给你几个针对性的优化方案，不用更换技术栈就能大幅提升速度：

1. 用Elasticsearch原生聚合替代Spark窗口函数（核心优化） #

你之前提到"Elasticsearch不支持与Spark结合的聚合操作"，其实elasticsearch-spark是支持通过提交ES聚合DSL来在ES端完成计算的——这样不需要把4亿条数据全量拉到Spark集群，只需要ES返回每个filename对应的首尾文档，数据量会骤降几个数量级。

具体实现思路： #

使用ES的terms聚合按filename分组，每个分组内用top_hits聚合分别获取timestamp最小和最大的文档（可以指定只返回你需要的5个字段）。如果filename的基数非常大（比如百万级以上），推荐用composite聚合代替terms，避免ES节点内存过载，因为composite支持分页返回聚合结果。

示例代码（Scala + elasticsearch-spark）： #

首先定义聚合DSL：

val aggDsl =
  """
    |{
    |  "size": 0,
    |  "aggs": {
    |    "filename_groups": {
    |      "terms": {
    |        "field": "filename",
    |        "size": 10000  // 根据你的filename基数调整，若基数极大改用composite
    |      },
    |      "aggs": {
    |        "earliest_doc": {
    |          "top_hits": {
    |            "size": 1,
    |            "sort": [{"timestamp": "asc"}],
    |            "_source": {"includes": ["systemname", "filename", "timestamp", "message", "version"]}
    |          }
    |        },
    |        "latest_doc": {
    |          "top_hits": {
    |            "size": 1,
    |            "sort": [{"timestamp": "desc"}],
    |            "_source": {"includes": ["systemname", "filename", "timestamp", "message", "version"]}
    |          }
    |        }
    |      }
    |    }
    |  }
    |}
  """.stripMargin

然后通过Spark读取ES聚合结果：

import org.elasticsearch.spark.sql._

val aggResult = spark.read
  .format("org.elasticsearch.spark.sql")
  .option("es.nodes", "your-es-nodes")
  .option("es.port", "9200")
  .option("es.query", aggDsl)
  .load("your-index-name/_search")

// 解析聚合结果，提取每个filename的首尾文档
import spark.implicits._
val finalResult = aggResult.select(
  $"key".alias("filename"),
  $"earliest_doc.hits.hits".getItem(0).getField("_source").alias("earliest"),
  $"latest_doc.hits.hits".getItem(0).getField("_source").alias("latest")
).select(
  $"filename",
  $"earliest.systemname", $"earliest.filename", $"earliest.timestamp", $"earliest.message", $"earliest.version",
  $"latest.systemname", $"latest.filename", $"latest.timestamp", $"latest.message", $"latest.version"
)

finalResult.show()

如果filename基数极大，把terms聚合换成composite：

val compositeAggDsl =
  """
    |{
    |  "size": 0,
    |  "aggs": {
    |    "filename_groups": {
    |      "composite": {
    |        "size": 10000,
    |        "sources": [{"filename": {"terms": {"field": "filename"}}}]
    |      },
    |      "aggs": {
    |        "earliest_doc": {
    |          "top_hits": {
    |            "size": 1,
    |            "sort": [{"timestamp": "asc"}],
    |            "_source": {"includes": ["systemname", "filename", "timestamp", "message", "version"]}
    |          }
    |        },
    |        "latest_doc": {
    |          "top_hits": {
    |            "size": 1,
    |            "sort": [{"timestamp": "desc"}],
    |            "_source": {"includes": ["systemname", "filename", "timestamp", "message", "version"]}
    |          }
    |        }
    |      }
    |    }
    |  }
    |}
  """.stripMargin

2. 强制减少ES返回的字段（解决你说的"无法仅查询特定字段"问题） #

elasticsearch-spark支持通过参数指定要包含的字段，你之前可能没配置正确。在读取ES数据时添加以下参数：

.option("es.read.field.include", "systemname,filename,timestamp,message,version")

这样ES只会返回你需要的5个字段，大幅降低网络I/O和Spark端的数据处理量。即使你暂时保留窗口函数的方案，这个参数也能有效提速。

3. 优化Elasticsearch索引的聚合性能 #

• 确保filename字段是keyword类型（如果之前是text类型，聚合会非常慢），timestamp是date类型。可以通过查看索引mapping确认：

  curl -XGET 'http://your-es-node:9200/your-index-name/_mapping'
  

  如果类型不对，需要重新建立索引并reindex（ES6.0支持reindex API）。
• 给timestamp字段保留正向索引（默认已经开启，若被禁用需重新启用），Min/Max和排序操作依赖索引快速定位。
• 确保ES集群有足够的堆内存（建议分配物理内存的50%，不超过32G），聚合操作对内存要求较高。

4. Spark端的辅助优化 #

• 调整shuffle分区数：设置spark.sql.shuffle.partitions为集群核数的2-3倍（112核的话设224-336），避免分区过大导致单个任务处理过慢，或分区过小导致任务调度开销大。
• 启用Kryo序列化：Spark默认用Java序列化，速度慢。在Spark配置中添加：

  spark.conf.set("spark.serializer", "org.apache.spark.serializer.KryoSerializer")
  spark.conf.set("spark.kryo.registrator", "org.elasticsearch.spark.serializer.EsKryoRegistrator")
  

• 合理分配Spark资源：112核集群可以设置每个Executor用8核，内存32G，共14个Executor，避免资源浪费。

优先顺序 #

建议按以下顺序尝试优化：

1. 先实现ES原生聚合方案，这是提升性能最显著的一步，能把数据量从4亿降到几千/几万条。
2. 配置字段过滤参数，减少I/O。
3. 检查并优化ES索引的mapping和内存配置。
4. 调整Spark的资源和序列化参数。

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