Spark SQL读写 ES7.x 及问题总结

本文主要介绍 spark SQL 读写 ES，参数的配置以及问题总结。

ES官方提供了对spark的支持，可以直接通过spark读写es，具体可以参考ES Spark Support文档

以下是pom依赖，具体版本可以根据自己的es和spark版本进行选择：

      <dependency>
        <groupId>org.elasticsearch</groupId>
        <artifactId>elasticsearch-spark-20_2.11</artifactId>
        <version>7.3.1</version>
      </dependency>

Spark SQL to ES

主要提供了两种读写方式：

一种是通过DataFrameReader/Writer传入ES Source实现；
另一种是直接读写DataFrame实现。

在实现前，还要列一些相关的配置：

参数	描述
es.nodes.wan.only	true or false，在此模式下，连接器禁用发现，并且所有操作通过声明的es.nodes连接
es.nodes	ES节点
es.port	ES端口
es.index.auto.create	true or false，是否自动创建index
es.resource	资源路径
es.mapping.id	es会为每个文档分配一个全局id。如果不指定此参数将随机生成；如果指定的话按指定的来
es.batch.size.bytes	es批量API的批量写入的大小（以字节为单位）
es.batch.write.refresh	批量更新完成后是否调用索引刷新
es.read.field.as.array.include	读es的时候，指定将哪些字段作为数组类型

列了一些常用的配置，更多配置查看ES Spark Configuration文档

DataFrameReader 读 ES

import org.elasticsearch.spark.sql._
val options = Map(
  "es.nodes.wan.only" -> "true",
  "es.nodes" -> "29.29.29.29:10008,29.29.29.29:10009",
  "es.port" -> "9200",
  "es.read.field.as.array.include" -> "arr1, arr2"
)
val df = spark
    .read
    .format("es")
    .options(options)
    .load("index1/info")
df.show()

DataFrameWriter 写 ES

import org.elasticsearch.spark.sql._
val options = Map(
  "es.index.auto.create" -> "true",
  "es.nodes.wan.only" -> "true",
  "es.nodes" -> "29.29.29.29:10008,29.29.29.29:10009",
  "es.port" -> "9200",
  "es.mapping.id" -> "id"
)

val sourceDF = spark.table("hive_table")
sourceDF
  .write
  .format("org.elasticsearch.spark.sql")
  .options(options)
  .mode(SaveMode.Append)
  .save("hive_table/docs")

读DataFrame

jar包中提供了 esDF() 方法可以直接读es数据为DataFrame，以下是源码截图。

参数说明：

resource：资源路径，例如index和tpye: hive_table/docs
cfg：一些es的配置，和上面代码中的options差不多
query：指定DSL查询语句来过滤要读的数据，例如"?q=user_group_id:3"表示读user_group_id为3的数据

val options = Map(
  "pushdown" -> "true",
  "es.nodes.wan.only" -> "true",
  "es.nodes" -> "29.29.29.29:10008,29.29.29.29:10009",
  "es.port" -> "9200"
)

val df = spark.esDF("hive_table/docs", "?q=user_group_id:3", options)
df.show()

写 DataFrame

jar包中提供了 saveToEs() 方法可以将DataFrame写入ES，以下是源码截图。

resource：资源路径，例如index和tpye: hive_table/docs
cfg：一些es的配置，和上面代码中的options差不多

示例：

val brandDF = sparkSession.sql(""" SELECT
              |   categoryname AS id
              | , concat_ws(',', collect_set(targetword)) AS targetWords
              | , get_utc_time() as `@timestamp`
              | FROM  t1
              | GROUP BY
              | categoryname
              """.stripMargin)

 // 手动指定ES _id值
 val map = Map("es.mapping.id" -> "id")
 EsSparkSQL.saveToEs(brandDF, "mkt_noresult_brand/mkt_noresult_brand", map)

Spark RDD to ES

SparkRDD方式写 ES，以下是源码截图。

示例：

    val numbers = Map("one" -> 1, "two" -> 2, "three" -> 3)
    val airports = Map("OTP" -> "Otopeni", "SFO" -> "San Fran")
    val rdd = sparkSession.sparkContext.makeRDD(Seq(numbers, airports))
    EsSpark.saveToEs(rdd, "mkt_noresult_brand/mkt_noresult_brand", map)

问题总结

手动指定ES _id值

EsSparkSQL.saveToEs 报错org.elasticsearch.hadoop.EsHadoopIllegalArgumentException: [DataFrameFieldExtractor for field [[...]]] cannot extract value from entity

原因："es.mapping.id"参数指定文档的id，这个参数必须配置成DataFrame中已有的字段，不能随意指定。配置成 val map = Map("es.mapping.id" -> "id"), 数据导入成功。