如何在Spark中读取多个同结构Elasticsearch索引并避免重复创建DataFrame？

嗨，这个场景太常见啦！既然你的索引结构完全一致，而且已经有了索引名数组，完全不用重复写多遍读取DataFrame的代码，用Scala的集合操作或者Elasticsearch的索引通配符就能轻松搞定，给你两种实用方案：

方案一：批量生成并合并DataFrame #

利用Scala的map遍历索引数组，为每个索引生成对应的DataFrame，再用reduce合并成一个统一的DataFrame，代码复用性拉满：

import org.apache.spark.sql.SparkSession

// 初始化SparkSession（如果还没初始化的话）
val spark = SparkSession.builder()
  .appName("ESMultiIndexRead")
  .getOrCreate()

val myquery = "你的Elasticsearch查询语句"
val indexNames = Array("news_01", "news_02")

// 批量处理索引并合并DataFrame
val combinedDF = indexNames.map { index =>
  spark.read.format("org.elasticsearch.spark.sql")
    .option("query", myquery)
    .option("pushdown", "true")
    .load(s"$index/myitem")
}.reduce(_ unionByName _)

这里用unionByName而不是普通的union，是为了保险起见——哪怕后续索引字段顺序有细微变化，也不会导致数据错位，毕竟你说结构完全一致，这个操作会更可靠。

方案二：使用Elasticsearch索引通配符（更高效） #

如果你的索引命名有规律（比如都是news_xx这种格式），直接用通配符*匹配多个索引，一次读取就能拿到所有数据，这比多次读取后合并更高效，因为是单次请求到ES集群：

val combinedDF = spark.read.format("org.elasticsearch.spark.sql")
  .option("query", myquery)
  .option("pushdown", "true")
  .load("news_*/myitem")

这个方案代码更简洁，性能也更好，优先推荐用这种方式，除非你的索引命名没有统一规律，那再用方案一。

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