如何使用Spark读取Elasticsearch中的嵌套数组数据？

嘿，我懂你在用Spark处理Elasticsearch数据时碰到数组类型字段的困扰——这确实是个挺常见的场景，我结合你的数据示例给你梳理几个实用的解决办法：

一、先确保Spark正确识别ES中的数组字段 #

首先，Spark和Elasticsearch的官方连接器（elasticsearch-hadoop）默认是能自动识别数组类型字段的，咱们先从基础的读取步骤确认：

import org.elasticsearch.spark._
import org.apache.spark.sql.SparkSession

val spark = SparkSession.builder()
  .appName("ESArrayDataProcessing")
  .config("es.nodes", "你的ES集群地址")
  .config("es.port", "9200")
  .getOrCreate()

// 读取目标ES索引的数据
val esRawDF = spark.read.format("org.elasticsearch.spark.sql")
  .option("es.resource", "Index_Name/Type_Name")
  .load()

// 打印Schema，确认数组字段的类型（比如Array[String]、Array[Struct]）
esRawDF.printSchema()

如果发现数组字段没被正确识别，先检查ES的索引映射，确保目标字段的type确实是array（比如"tags": {"type": "array", "items": "string"}）。

二、最常用：把数组展开为单独行 #

如果你的需求是将数组中的每个元素拆成独立的数据行（比如统计每个数组元素的出现次数），用Spark的explode函数就搞定了：

示例1：展开普通数组 #

假设你的文档里有个tags数组字段（比如["shop", "restaurant", "park"]）：

import org.apache.spark.sql.functions._

val explodedDF = esRawDF.select(
  col("currentTime"),
  col("location.lat").alias("lat"),
  col("location.lon").alias("lon"),
  // 展开数组字段，每个元素生成一行
  explode(col("tags")).alias("single_tag")
)

explodedDF.show(10)

示例2：展开嵌套数组（数组内是结构体） #

如果数组里是嵌套的结构体（比如events: [{"action": "click", "timestamp": 1518339120000}, ...]），可以先展开再提取结构体字段：

val nestedExplodedDF = esRawDF.select(
  col("currentTime"),
  col("location.*"),
  explode(col("events")).alias("event_detail")
).select(
  col("currentTime"),
  col("lat"),
  col("lon"),
  col("event_detail.action").alias("action_type"),
  col("event_detail.timestamp").alias("action_time")
)

要是数组可能为空或null，不想过滤掉这些行，就用explode_outer代替explode。

三、不展开数组：直接对数组做聚合操作 #

如果不需要拆分行，只是想对数组本身做计算（比如统计长度、去重、求和），用Spark的数组函数就能实现：

// 统计数组元素个数
val arrayLenDF = esRawDF.withColumn("tags_count", size(col("tags")))

// 数组元素去重
val distinctTagsDF = esRawDF.withColumn("unique_tags", array_distinct(col("tags")))

// 数值型数组求和（比如你的radius如果是数值数组）
val sumRadiusDF = esRawDF.withColumn("total_radius", aggregate(col("radius"), lit(0), (acc, x) => acc + x))

四、处理后的数据写回Elasticsearch #

如果要把处理好的结果写回ES，Spark连接器会自动把DataFrame中的数组类型字段映射为ES的数组类型，只要配置好基本参数即可：

nestedExplodedDF.write.format("org.elasticsearch.spark.sql")
  .option("es.resource", "processed_index/processed_type")
  .option("es.mapping.id", "_id") // 可选：指定用原文档的ID作为新文档ID
  .mode("append")
  .save()

常见问题排查 #

• 数组被识别成单个值：先检查ES的索引映射，确认字段是array类型；也可以用array_contains函数测试：esRawDF.filter(array_contains(col("tags"), "shop")).show()，如果能查到结果，说明是数组类型。
• 展开后丢失null/空数组的行：替换explode为explode_outer，保留所有原始行。

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