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默认保留时间scroll1m到5m期间这些上下文占着堆内存如果并发游标太多ES 的堆会被吃掉一大块。还有第三种search_after它基于排序值翻页不需要维护游标上下文但要求有稳定的排序字段且只能顺序翻页。Spark 的官方连接器在较新版本里对search_after也有支持但主流还是走 scroll 或分片切片。理解这三种出口是理解 Spark 各方案的前提。因为 Spark 连接器本质上就是用某种方式把 ES 的数据出口包一层映射成 Spark 的 Partition 和 Iterator。2.2 Spark 侧的分区模型为什么分区数决定了并行度上限Spark 读任何数据源最终都要落到 RDD/DataFrame 的 Partition 上。一个 Partition 就是一个 TaskTask 数决定了并行度。读 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 返回的是 JSONSpark 连接器要把它反序列化成 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 四种方案的对比表方案并行度来源适用数据规模对 ES 压力一致性实现复杂度默认模式分片数粗 10GB低快照极低slice scrollslice 数10GB ~ 数 TB中高可控快照低手动范围切分切分段数有均匀分区键时任意中无统一快照中导出列式再读文件分区反复分析场景一次性取决于导出高这张表是我做选型时的核心依据。可以看到没有一种方案在所有维度上都赢选型就是根据你的数据规模、分析频次、一致性要求去挑最合适的那个。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设得太小往返次数爆炸设得太大单次响应可能几十 MBES 协调节点要把它拼出来内存压力大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 推断、字段映射、连接器参数都生效了再放开全量。这一步花两分钟能省掉后面几十分钟的无效等待。数据工程这行验证永远比蛮干划算。