如何优化Spark集群从Elasticsearch并行加载海量数据？

看起来你遇到的核心问题是Elasticsearch-Hadoop连接器没有正确将ES分片映射为Spark分区，导致所有读取逻辑都落在Driver上，完全没用到Executor的并行能力。下面是针对你的场景的具体优化方案：

1. 让Spark自动感知ES集群的分片结构 #

你的代码只指定了单个ES节点IP，这会让连接器无法自动发现整个集群的分片分布。添加分片发现配置，让Spark知晓每个ES分片的位置，从而分配Executor并行读取：

df = sqlContext.read.format("org.elasticsearch.spark.sql")\
    .option("es.nodes", es_ip)\
    .option("es.nodes.discovery", "true")  # 开启自动发现ES集群节点与分片
    .option("es.nodes.wan.only", "true")  # 跨网络访问ES（如云环境）时需添加
    .load(es_index)

开启es.nodes.discovery后，连接器会从指定节点获取整个ES集群的元数据，包括所有主分片的分布，这样Spark就能为每个ES主分片创建对应的读取任务，分配给不同Executor执行。

2. 拆分ES分片为更多Spark分区，拉满并行度 #

你的ES索引有5个主分片，默认情况下连接器只会生成5个Spark分区，和你设置的16个Executor不匹配，没法充分利用资源。可以通过es.input.split.size参数，将单个ES分片拆分为多个Spark分区：

.option("es.input.split.size", "200000")  # 每个Spark分区从ES分片读取最多20万条文档

这个值可根据单条文档大小调整：文档小（几KB）可设到50万，文档大（几十KB）设为10万更合适。这样每个ES主分片会被拆分成多个子任务，让更多Executor参与读取。

3. 优化Spark作业资源配置，匹配集群硬件 #

你的集群有3个节点，每个8核16GB，总资源为24核48GB。当前16个1核2GB的Executor配置没有充分利用硬件，建议调整为：

# 提交作业时的配置
spark-submit \
  --master spark://master:7077 \
  --executor-cores 2 \
  --executor-memory 4g \
  --num-executors 12 \
  --driver-memory 4g \
  your_script.py

每个节点可跑4个2核4GB的Executor（8核/2核=4，16GB/4GB=4），3个节点刚好12个Executor，把集群资源用满，并行能力拉满。

4. 调整分区操作时机，避免限制读取并行度 #

你的代码先执行coalesce(16)再写入，会导致读取阶段的并行度被限制。正确做法是：

• 先让连接器生成足够的并行分区完成读取
• 之后再根据写入需求调整分区数

# 先并行读取
df = sqlContext.read.format("org.elasticsearch.spark.sql")\
    .option("es.nodes", es_ip)\
    .option("es.nodes.discovery", "true")\
    .option("es.input.split.size", "200000")\
    .load(es_index)

# 写入前调整分区：原分区数大于16用coalesce（无shuffle），小于16用repartition（触发shuffle增加分区）
df.repartition(16).write.option("compression","gzip").parquet(pq_filename)

5. 辅助优化：提升批量读取效率 #

添加ES批量读取配置，优化读取性能：

.option("es.batch.size.entries", "10000")\  # 单次批量读取1万条文档
.option("es.batch.size.bytes", "10mb")\     # 单次批量读取不超过10MB
.option("es.read.field.as.array.include", "your_array_fields")  # 提前声明数组类型字段，避免解析错误

这些配置可避免单次读取过多数据导致的内存压力，同时提升ES与Spark之间的数据传输效率。

按照以上步骤调整后，Spark会自动将读取任务分配到多个Executor并行执行，加载时间会大幅缩短（从8小时降到几十分钟甚至更短，取决于数据大小和集群性能）。

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