scala - 在 SPARK 中使用 elasticsearch-spark 从 Elasticsearch 读取数据时如何转换类型

Question

当我尝试使用 elasticsearch-spark 中的 esRDD("index") 函数从 elasticsearch 读取数据时，我得到的结果类型为 org.apache.spark.rdd.RDD [(String, scala.collection.Map[String,AnyRef])]。当我检查这些值时，它们都是 AnyRef 类型。但是，我在 ES site 上看到了，它说:

elasticsearch-hadoop automatically converts Spark built-in types to Elasticsearch types (and back)

我的依赖项是:

scalaVersion := "2.11.8"
libraryDependencies += "org.apache.spark" %% "spark-core" % "2.1.0"  
libraryDependencies += "org.apache.spark" %% "spark-sql" % "2.1.0"  
libraryDependencies += "org.apache.spark" %% "spark-mllib" % "2.1.0"  
libraryDependencies += "org.elasticsearch" % "elasticsearch-spark-20_2.11" % "5.4.0"

我错过了什么吗？以及如何以方便的方式转换类型？

Answer 1

好的，我找到了解决方案。如果您使用 esRDD，所有类型信息都会丢失。
如果我们使用:

val df = sparkSession.read.format("org.elasticsearch.spark.sql").option("es.read.field.as.array.include", "").load("index")

可以在option中配置es，如果之前配置过，可以忽略option。

数据返回在DataFrame中，数据类型在schema中保留(转换为sql.DataTypes)只要支持转换elasticsearch-spark.

现在您可以随心所欲了。