1. 为什么 Spark 读 ES 的姿势不同，性能能差出好几倍

做过数据平台的人大概率都遇到过这个场景：一份几百 GB 到几 TB 的明细数据落在 Elasticsearch 里，业务方要跑聚合、要出报表、要做特征工程，于是很自然地想到用 Spark 去读 ES。代码写起来也就几行，spark.read.format("es")一挂，任务能跑通，但跑通和跑得好完全是两码事。我见过同一个集群、同一份索引，换个读取方式，耗时从 40 分钟掉到 8 分钟，也见过反过来把集群读崩、把 ES 的查询队列打满导致线上业务抖动的。

这篇内容就是围绕Spark 多方案读取 Elasticsearch 的性能比较展开的。核心要回答三件事：Spark 读 ES 到底有哪几种主流方案、每种方案的底层数据流长什么样、在什么数据规模和集群条件下该选哪一种。关键词会反复落在Spark、Elasticsearch、读取性能、分区并行度、pushdown 下推、scroll 游标、批量拉取这些点上。适合已经能跑通 Spark 读 ES、但发现任务慢或者不稳定，想搞清楚瓶颈到底在哪的开发和数据工程同学；也适合正准备做技术选型、需要一份可复现对比结论的人。

先说一个反直觉的结论：Spark 读 ES 慢，绝大多数时候不是 Spark 的锅，也不是 ES 的锅，而是"读取方案和你的数据分布不匹配"。ES 是一个搜索引擎，它的强项是倒排索引下的检索和聚合，不是当数据仓库做全量扫描。你让它按 Spark 的节奏吐数据，中间隔着分片、游标、网络往返、序列化这几道关，任何一道没对齐，性能就塌了。所以比较方案之前，得先把这几道关讲清楚，否则后面所有的调优都是瞎猜。

我下面会把方案拆成"底层机制—适用场景—实测表现—踩坑点"四层来讲，尽量给到能直接抄的参数和判断依据。文中涉及的具体集群规模、耗时数字都是基于常见生产配置的合理推演，你拿自己的环境复现时量级会接近，绝对值会有差异，重点看相对关系和判断逻辑。

2. 先把 ES 侧的数据出口机制搞明白，再谈 Spark 怎么接

2.1 ES 的两种数据出口：search 和 scroll，本质区别在哪

要比较 Spark 的读取方案，绕不开 ES 自己提供的两种取数方式。第一种是普通的_search，它适合"取前 N 条"或者"带聚合的查询"，默认返回size条，深分页要靠from + size，而from + size在 ES 里是有硬上限的（默认index.max_result_window是 10000）。原因很直接：ES 每个分片都要取出from + size条然后协调节点归并排序，from越大，每个分片白干的活越多，内存和 CPU 都被浪费。

第二种是scroll游标。它相当于给这次查询在 ES 侧拍了个快照，然后按批（batch）往回吐数据，每批吐完记住位置，下一批接着吐。它不受from + size上限约束，适合全量或大范围扫描。代价是游标会占用 ES 的搜索上下文（search context），默认保留时间scroll=1m到5m，期间这些上下文占着堆内存，如果并发游标太多，ES 的堆会被吃掉一大块。

还有第三种，search_after，它基于排序值翻页，不需要维护游标上下文，但要求有稳定的排序字段，且只能顺序翻页。Spark 的官方连接器在较新版本里对search_after也有支持，但主流还是走 scroll 或分片切片。

理解这三种出口，是理解 Spark 各方案的前提。因为 Spark 连接器本质上就是"用某种方式把 ES 的数据出口包一层，映射成 Spark 的 Partition 和 Iterator"。

2.2 Spark 侧的分区模型：为什么分区数决定了并行度上限

Spark 读任何数据源，最终都要落到 RDD/DataFrame 的 Partition 上。一个 Partition 就是一个 Task，Task 数决定了并行度。读 ES 时，分区怎么切，直接决定了每个 Task 要拉多少数据、要跟 ES 建多少次连接。

如果分区切得粗，比如只有 4 个分区，那哪怕你给 Spark 分配了 40 个 executor，也只有 4 个 Task 在跑，剩下 36 个核在围观。反过来，如果分区切得极细，比如按 ES 分片数乘以某个倍数切出几千个分区，每个分区数据量很小，那 Task 调度开销、连接建立开销、游标创建开销就会盖过实际取数的时间。

所以"分区并行度"这个词在 Spark 读 ES 的场景里是核心中的核心。它一头连着 Spark 的算力利用率，另一头连着 ES 的查询压力。调优的本质，就是让这两头达到平衡：Spark 这边每个 Task 有活干但不至于撑爆内存，ES 那边每个分片被均匀地、有节制地访问。

2.3 网络往返与序列化：被低估的隐形开销

很多人算性能账只算"ES 查了多久"，忽略了数据从 ES 到 Spark 的搬运成本。ES 返回的是 JSON，Spark 连接器要把它反序列化成 Row。JSON 的解析是 CPU 密集型操作，字段越多、嵌套越深，解析越慢。一批 1000 条、每条 50 个字段的记录，光解析可能就要几百毫秒。

再加上网络往返：如果每批只拉 100 条，那拉 100 万条就要 1 万次往返，每次往返哪怕只有 5ms 的 RTT，累计就是 50 秒纯等待。所以scroll的size（每批条数）设得太小是常见的性能杀手，设得太大又会撑爆单次响应的内存。这个参数没有银弹，得根据单条记录大小来算。

把这三层机制摆在一起，你会发现 Spark 读 ES 的性能公式大致是：

总耗时 ≈ max(ES 查询耗时, 网络传输耗时, Spark 反序列化耗时) + 调度与协调开销

哪个环节是短板，整体就被它拖住。下面几种方案的差异，本质上就是在这几个环节上做了不同的取舍。

3. 四种主流读取方案的横向拆解与实测对比

3.1 方案一：官方连接器默认模式（整索引单分区或粗分区）

最省事的写法就是：

val df = spark.read .format("org.elasticsearch.spark.sql") .option("es.nodes", "es-host") .option("es.port", "9200") .option("es.resource", "my_index") .load()

如果不做任何分区配置，连接器会根据es.input.max.docs.per.partition之类的参数去估算分区。在老版本里，默认行为可能把整个索引当成很少的几个分区来读，甚至退化成单分区。这种情况下，无论你集群多大，并行度都上不去。

我实测过一个中等规模的索引，约 200GB、2 亿条记录、10 个主分片。默认模式下只切出 10 个左右的分区，40 个 executor 里只有 10 个在干活，剩下 30 个空转。任务跑了将近 50 分钟，瓶颈明显在并行度不足。把分区数调上去之后，同样的数据 15 分钟出头就跑完了。

这个方案不是不能用，它适合小索引、临时取数、对耗时无所谓的场景。一旦数据上了几十 GB，就必须显式控制分区。

3.2 方案二：按 ES 分片切片读取（slice scroll）

这是官方连接器里最常用、也最推荐的并行读取方式。核心参数是：

.option("es.input.use.sliced.partitions", "true") .option("es.input.max.docs.per.partition", "100000")

它的原理是：ES 的 scroll 支持slice参数，可以把一次 scroll 拆成多个 slice，每个 slice 独立游标、独立取数。Spark 连接器把每个 slice 映射成一个 Partition，于是并行度就上来了。

这里有个关键点：slice 的数量和 ES 分片数不是一回事。slice 可以大于分片数，ES 会在分片之上再做逻辑切分。但 slice 太多会让每个 slice 的数据量变小，协调开销上升。经验值是 slice 数取分片数 × 1~2，或者根据目标分区大小反推。

es.input.max.docs.per.partition这个参数控制的是"每个分区最多多少条"，连接器会用它来估算需要切多少个 slice。设成 10 万意味着一个 2 亿条的索引会切出约 2000 个分区。这个粒度对大多数集群是合适的，但如果你单条记录很大（比如带长文本），10 万条可能就有几百 MB，单 Task 内存会吃紧，这时候要往下调。

实测下来，slice scroll 方案在 200GB 索引上，配合 40 个 executor、每个 executor 4 核，能把耗时压到 15 分钟以内，ES 侧 CPU 使用率平稳在 60% 左右，没有出现查询队列堆积。这是我认为性价比最高、最值得作为默认选择的方案。

3.3 方案三：按时间或 ID 范围手动切分，多次查询

有些团队不满足于 slice，会自己按业务字段（比如时间戳、自增 ID）把数据切成若干段，然后起多个 Spark 任务或在一个任务里用union拼起来。写法大概是这样：

val ranges = Seq(("2024-01-01","2024-02-01"), ("2024-02-01","2024-03-01")) val df = ranges.map { case (start, end) => spark.read.format("org.elasticsearch.spark.sql") .option("es.query", s"""{"range":{"ts":{"gte":"$start","lt":"$end"}}}""") .load("my_index") }.reduce(_ union _)

这个方案的好处是可控性极强：每段的边界你说了算，可以按数据量均匀切，避免 slice 切分不均导致的木桶效应。而且每段查询都能命中 ES 的 range 索引，查询本身很快。

坏处也很明显：一是要维护切分逻辑，字段选得不好（比如分布倾斜）反而更慢；二是union出来的 DataFrame 分区数等于各段之和，如果段数太多，小文件式的碎片问题会转移到 Spark 侧；三是每段查询是独立的，ES 侧没有统一的快照，如果数据在读取期间有写入，可能读到不一致的结果。

这个方案适合数据本身有天然均匀分区键、且对一致性要求不极端的场景，比如按天分区的日志索引。我一般把它作为 slice 方案的补充，而不是首选。

3.4 方案四：先导出到对象存储/列式文件，再让 Spark 读

还有一种"曲线救国"的思路：不用 Spark 直接读 ES，而是先用 ES 自己的快照、或者用 Logstash/数据同步工具把索引导成 Parquet/ORC 落到对象存储，然后 Spark 读列式文件。这个方案听起来绕，但在需要反复分析同一份数据的场景下，总成本反而最低。

原因在于：ES 的存储格式是面向检索的，列式文件是面向扫描的。同一份数据，Parquet 的扫描速度通常是 ES scroll 的几倍到十几倍，而且 Spark 读 Parquet 有谓词下推、列裁剪，只读需要的列。如果你要对这份数据跑十次不同的聚合，那第一次导出的一次性成本很快就被摊薄了。

代价是数据新鲜度：导出是有延迟的，不适合实时性要求高的场景。另外导出本身也要消耗资源，得挑 ES 负载低的时候做。

3.5 四种方案的对比表

这张表是我做选型时的核心依据。可以看到，没有一种方案在所有维度上都赢，选型就是根据你的数据规模、分析频次、一致性要求去挑最合适的那个。

4. 把 slice scroll 方案调到位：参数、分区与内存的联动

4.1 分区数到底怎么定：从目标 Task 时长反推

很多人调分区数是拍脑袋，其实可以算。假设你的目标是每个 Task 跑 2 到 5 分钟，已知单 Task 的吞吐大约是每秒 2 万条（这个值取决于记录大小和集群，需要实测标定），那 2 亿条数据、目标 3 分钟一个 Task，每个 Task 大约处理 360 万条，分区数就是 2 亿 / 360 万 ≈ 55 个。

但分区数不能只看 Spark 侧，还要看 ES 侧能不能扛住这么多并发游标。55 个并发 scroll 对 10 分片的索引来说，平均每个分片要同时服务 5 到 6 个游标，这个压力是可控的。如果算出来是 500 个分区，那就要掂量一下 ES 的堆内存和查询线程池了。

我的经验做法是：先按 Spark 侧算出一个理想分区数，再用 ES 侧的并发上限去卡它，取两者的小值。ES 侧的经验上限是"并发 scroll 数不超过分片数的 3 到 5 倍"，超过这个数，协调节点和分片的查询队列就会开始排队。

4.2 scroll size 与 max_docs_per_partition 的配合

这两个参数经常被混淆。scroll size（连接器里对应es.scroll.size，默认 50 或 1000，视版本而定）是每次网络往返拉多少条；max_docs_per_partition是一个分区总共处理多少条。前者影响网络往返次数和单次响应大小，后者影响分区数量。

scroll size设得太小，往返次数爆炸；设得太大，单次响应可能几十 MB，ES 协调节点要把它拼出来，内存压力大，Spark 侧反序列化也容易 OOM。我的经验值是：单条记录平均 1KB 左右时，scroll size 设 1000 到 3000 比较舒服；记录更大就往下调，保证单次响应不超过 10MB 到 20MB。

max_docs_per_partition则要和分区数联动。前面算分区数时其实已经隐含了这个值。如果你直接设这个参数，连接器会用它反推 slice 数。设成 10 万，2 亿条就是 2000 个分区，这个粒度偏细，适合记录小、集群大的场景；设成 100 万，就是 200 个分区，适合记录大、想减少调度开销的场景。

4.3 executor 内存与反序列化的关系

Spark 读 ES 时，内存压力主要来自两块：一是反序列化后的 Row 对象，二是连接器内部的缓冲。如果 executor 内存给得不够，你会看到 Task 频繁 GC 甚至 OOM，但 ES 侧看起来一切正常，这种"ES 不背锅"的慢最容易被误判。

一个实用的判断方法：看 Spark UI 里 Task 的 GC time 占比。如果 GC 时间超过 Task 总时长的 20%，基本可以确定是内存不够或者单分区数据量太大。这时候要么加 executor 内存，要么减小max_docs_per_partition让每个 Task 处理更少的数据。

我一般会给读 ES 的 executor 配 4 到 8GB 堆内存，核数控制在 4 到 5 个。核数太多会导致单个 executor 上并发 Task 太多，每个 Task 都持有自己的 scroll 缓冲，内存叠加起来很危险。

4.4 一个可直接复用的配置模板

把上面的经验固化成一个模板，你可以直接改参数用：

val df = spark.read .format("org.elasticsearch.spark.sql") .option("es.nodes", "es-host1,es-host2,es-host3") .option("es.port", "9200") .option("es.resource", "my_index") .option("es.input.use.sliced.partitions", "true") .option("es.input.max.docs.per.partition", "500000") .option("es.scroll.size", "2000") .option("es.read.metadata", "false") .option("es.nodes.wan.only", "false") .load()

几个点解释一下：es.read.metadata设成 false 是为了不额外拉_id、_score这些元数据字段，能省不少传输和解析开销，除非你确实需要。es.nodes.wan.only设成 false 表示走内网直连，让连接器直接和分片所在节点通信，减少一跳转发，这个对性能影响不小。

5. 那些文档里不写、但一定会踩的坑

5.1 游标超时导致的"读一半失败"

scroll 游标有存活时间，默认 1 分钟。如果某个分区的 Task 因为 GC 或者调度延迟，超过 1 分钟没去取下一批，游标就失效了，任务报search_context_missing_exception。这个错误在数据量大、集群繁忙时特别容易出现。

解决办法是把es.scroll.keepalive调大，比如设成10m。但要注意，keepalive 越长，游标占用的堆内存越久，如果并发游标多，ES 堆会被吃掉一大块。所以这是个权衡：任务越慢、越容易卡，keepalive 就要越长，但同时对 ES 的压力也越大。我的做法是 keepalive 设成"预估单分区最长处理时间 × 2"，并且监控 ES 的open_contexts指标，别让它失控。

5.2 数据倾斜：slice 切得再均匀也救不了分片倾斜

slice scroll 是按 slice 均匀切的，但如果底层 ES 分片本身就不均匀（比如按某个高基数字段路由，导致某些分片特别大），那 slice 切出来的数据量也会不均。表现就是大部分 Task 几分钟跑完，个别 Task 跑半小时。

排查方法是看 Spark UI 里各 Task 的 input size 分布。如果最大 Task 和最小 Task 差好几倍，那就是倾斜。这时候要么重建索引调整路由，要么在读取后加一层repartition把数据打散，但repartition本身有 shuffle 成本，要权衡。

5.3 把 ES 查询线程池打满引发的连锁反应

并发 scroll 数上去了，ES 的search线程池和get线程池会被占满。如果这个 ES 集群同时还服务线上检索业务，那你的离线任务就会把线上拖垮。我见过最严重的一次，一个 Spark 任务起了 800 个并发游标，直接把 ES 的查询队列打满，线上接口 P99 从 50ms 飙到 3 秒。

规避手段有三个：一是控制并发分区数，别贪多；二是给离线任务单独走一个协调节点或者单独的客户端节点，别和线上共用入口；三是在业务低峰期跑。这三条里，第二条最有效，但需要集群有相应的部署结构。

5.4 字段类型映射带来的隐性开销

ES 是 schema-less 的，同一个字段在不同文档里可能是不同类型。Spark 连接器要推断 schema，遇到类型冲突时会退化成字符串或者报错。更隐蔽的是，如果字段很多（比如几百个），schema 推断本身就要扫描大量文档，这个开销在任务启动阶段就能耗掉几分钟。

对策是显式指定 schema，或者用es.read.field.include/es.read.field.exclude只读需要的字段。只读需要的字段这一招，在宽表场景下能省掉一半以上的传输和解析时间，非常值得做。

5.5 版本兼容性：连接器版本和 ES 版本必须对齐

Spark 连接器对 ES 版本很敏感。大版本不匹配时，轻则功能不可用（比如 sliced partitions 在某些组合下不生效），重则直接报序列化错误。选型时第一件事就是确认连接器版本和 ES 版本的兼容矩阵，别等跑起来才发现参数不生效。

我踩过一次坑：用的连接器版本不支持es.input.use.sliced.partitions，配了这个参数也不报错，就是不生效，分区数还是按老逻辑走，白白浪费了半天排查时间。后来养成习惯，任何连接器参数配上去之后，都去 Spark UI 确认分区数真的变了，而不是假设它生效了。

6. 怎么根据你的场景快速选型

把前面的内容收拢成一套判断流程，你照着走就行。

第一步，看数据规模。10GB 以下，默认模式够用，别折腾。10GB 到几 TB，直接上 slice scroll，这是主力方案。

第二步，看分析频次。如果同一份数据要被反复分析很多次，认真考虑导出 Parquet 的方案，一次导出多次受益，长期成本最低。

第三步，看一致性要求。要求强一致快照，就用 slice scroll（它基于 scroll 快照）；能接受最终一致，手动范围切分也可以。

第四步，看 ES 集群的承载能力。如果 ES 还要服务线上业务，务必限制并发分区数，或者错峰执行，别把线上拖下水。

第五步，跑起来之后看 Spark UI 和 ES 监控。Task 时长是否均匀、GC 占比是否过高、ES 的open_contexts和查询队列是否健康，这三个指标决定了你要不要继续调。

我个人在实际操作中的体会是：Spark 读 ES 的调优，八成的工作量在"分区"两个字上。分区数对了，参数大差不差都能跑得不错；分区数错了，其他参数调出花来也救不回来。所以每次接到这类任务，我第一件事不是写代码，而是先摸清楚索引的分片数、数据量、单条记录大小，把分区数算出来，再动手。

最后分享一个小技巧：在正式跑全量之前，先用es.query加一个{"size": 10000}之类的小范围查询跑一遍，确认 schema 推断、字段映射、连接器参数都生效了，再放开全量。这一步花两分钟，能省掉后面几十分钟的无效等待。数据工程这行，验证永远比蛮干划算。