调优Elasticsearch写入性能：Spark批量写入慢问题排查

优化方案与排查点 #

一、Elasticsearch集群优化

1. 分片与副本配置
   • 最优分片数参考：节点数 × CPU核心数/2，5节点×8=40分片左右（16核节点单节点活跃分片建议不超8个）。避免分片过少（单分片压力过大）或过多（元数据开销高）。
   • 写入阶段临时关闭副本：提前创建索引并设置index.number_of_replicas: 0，写入完成后再恢复目标副本数，消除副本同步开销。
   • 预创建索引：不要依赖Spark自动创建索引，提前定义好字段映射与分片参数，避免写入时的初始化延迟。
2. ES核心参数调优
   修改elasticsearch.yml：
   • indices.memory.index_buffer_size: 30%（默认10%，提升索引缓冲内存占比）
   • thread_pool.write.size: 8（设置为CPU核心数的一半）
   • thread_pool.write.queue_size: 1000（增大写入队列，避免请求被拒绝）
   • 写入期间临时关闭刷新：设置index.refresh_interval: -1，完成后恢复为30s左右；同时调整合并线程数index.number_of_merge_threads: 8，提升段合并效率。
3. 磁盘IO优化
   • 优先使用SSD磁盘，机械盘IO瓶颈会严重限制写入速度。
   • 调整磁盘限流：SSD环境下可将indices.store.throttle.max_bytes_per_sec调至500mb，避免段合并时IO被限流。

二、Spark写入ES配置优化

1. 并行度与批量参数调整
   • 调整DataFrame分区数：9.15亿行建议分成450-900个分区（单分区100-200万行），确保并行度匹配Spark集群资源（至少为CPU核数的2-3倍）。可通过whole_df.repartition(500)调整。
   • 优化ES写入参数，修改writeDFToEs函数：

     def writeDFToEs(df: DataFrame, index: String) = {
         df.write
           .format("org.elasticsearch.spark.sql")
           .option("es.nodes", "192.168.1.xxx")
           .option("es.http.timeout", 600000)
           .option("es.http.max_content_length", "2000mb")
           .option("es.port", 9200)
           .option("es.batch.size.entries", 10000) // 单批量条目数，从默认1000调至1万
           .option("es.batch.size.bytes", "15mb") // 单批量大小，从默认5mb调至15mb
           .option("es.batch.write.refresh", "false") // 写入阶段禁止自动刷新
           .option("es.batch.write.retry.count", 5) // 失败重试次数
           .option("es.batch.write.retry.wait", "10s") // 重试间隔
           .option("es.nodes.wan.only", "true") // 跨网段访问时开启
           .mode("overwrite")
           .save(s"$index")
     }
     

   • Spark Executor配置：每个Executor分配10核、40GB内存（10节点×15核可拆分出15个Executor），减少GC频率。提交任务时添加：

     --executor-cores 10 --executor-memory 40g --driver-memory 16g
     

2. 序列化优化
   • 使用Kryo序列化替代默认Java序列化，提交Spark任务时添加：

     --conf spark.serializer=org.apache.spark.serializer.KryoSerializer
     --conf spark.kryo.registrationRequired=false
     

   • 确保DataFrame字段类型与ES索引映射完全匹配，避免自动类型转换开销。

三、数据预处理优化

1. Parquet文件读取优化
   • 合并S3上的小Parquet文件至128-256MB/个，减少Spark读取时的小文件开销。
   • 添加S3读取优化配置：

     --conf spark.hadoop.fs.s3a.readahead.range=256mb
     --conf spark.hadoop.fs.s3a.block.size=128mb
     --conf spark.hadoop.fs.s3a.connection.maximum=100
     

2. 字段过滤
   • 提前用df.select(...)过滤无需写入ES的字段，减少数据传输量。

四、排查验证

1. ES集群监控
   • 查看节点资源使用率：curl -XGET 'http://192.168.1.xxx:9200/_cat/nodes?v'，若CPU持续90%+，需调整分片数或批量参数；若磁盘IO满负载，优先更换SSD。
   • 查看写入队列状态：curl -XGET 'http://192.168.1.xxx:9200/_cat/thread_pool/write?v'，队列持续满则需增大队列或提升ES处理能力。
2. Spark任务监控
   • 通过Spark UI查看Stage执行情况：排查数据倾斜（单任务耗时远超平均）、GC占比（超过20%则调整内存/序列化）。

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