spark读取elasticsearch中数组类型的字段

Spark可以使用Elasticsearch-Hadoop库来读取Elasticsearch中的数据。对于数组类型的字段，可以使用Spark SQL中的explode函数将其展开为多行数据。

例如，假设Elasticsearch中有一个名为“tags”的数组类型字段，可以使用以下代码将其读取为Spark DataFrame：

import org.apache.spark.sql.SparkSession

val spark = SparkSession.builder()
  .appName("Read Elasticsearch")
  .master("local[*]")
  .config("es.nodes", "localhost")
  .config("es.port", "9200")
  .config("es.index.auto.create", "true")
  .getOrCreate()

val df = spark.read.format("org.elasticsearch.spark.sql")
  .option("es.query", "?q=*")
  .load("my_index/my_type")

df.printSchema()

然后，可以使用explode函数将“tags”字段展开为多行数据：

import org.apache.spark.sql.functions.explode

val explodedDf = df.select(explode($"tags").as("tag"))
explodedDf.show()

这将展开“tags”字段，并将其作为新列“tag”添加到DataFrame中。现在，每个文档将变为多行，其中每个行都包含一个“tag”值。