【spark工具库】spark sql读写es操作

淩亂°似流年 2022-11-11 05:48 261阅读 0赞

> 本文主要介绍spark sql读写es、structured streaming写入es以及一些参数的配置
> 
> 文章转自：[https://www.cnblogs.com/upupfeng/p/12205657.html][https_www.cnblogs.com_upupfeng_p_12205657.html]

ES官方提供了对spark的支持，可以直接通过spark读写es，具体可以参考ES Spark Support[文档][Link 1](文末有地址)。

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

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

## Spark SQL - ES ##

主要提供了两种读写方式：一种是通过DataFrameReader/Writer传入ES Source实现；另一种是直接读写DataFrame实现。在实现前，还要列一些相关的配置：

### 配置 ###

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

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

### 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，以下是源码截图。  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2lmZW5nZ2VnZQ_size_16_color_FFFFFF_t_70]  
简单说一下各个参数：

resource：资源路径，例如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，以下是源码截图。  
![在这里插入图片描述][20200117142726916.png]  
resource：资源路径，例如hive\_table/docs

cfg：一些es的配置，和上面代码中的options差不多

## Structured Streaming - ES ##

es也提供了对Structured Streaming的集成，使用Structured Streaming可以实时的写入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" -> "zip_record_id"
    )
    val df = spark
      .readStream
      .format("kafka")
      .option("kafka.bootstrap.servers", "a:9092,b:9092,c:9092")
      .option("subscribe", "test")
      .option("failOnDataLoss", "false")
      .load()
    df
      .writeStream
      .outputMode(OutputMode.Append())
      .format("es")
      .option("checkpointLocation", s"hdfs://hadoop:8020/checkpoint/test01")
      .options(options)
      .start("test_streaming/docs")
      .awaitTermination()

## 可能遇到的问题 ##

### 数组类型转换错误 ###

报错信息：type (scala.collection.convert.Wrappers.JListWrapper) cannot be converted to the string type

因为es的mapping只会记录字段的类型，不会记录是否是数组，也就是说如果是int数组，es的mapping只是记录成int。

可以在option中加一个es.read.field.as.array.include，标明数组字段

`es.read.field.as.array.include" -> "数组字段的名字"`

如果是object里的某个字段，写成"object名字.数组字段名字"，如果是多个字段，字段名之间用逗号分隔

### Timestamp被转为Long ###

DataFrame的Timestamp类型数据写入ES后，就变成了Number类型。

这可能不算个问题，时间戳本质上就是Long类型的毫秒值；但是在Hive中Timestamp是"yyyy-MM-dd HH:mm:ss"的类型，个人觉得很别扭。

尝试将Timestamp类型字段转成Date类型，写入ES后还是Number类型。网上搜了一圈也没有什么好的办法，大家有什么解决办法欢迎交流。

## References ##

ES Spark Support文档：[https://www.elastic.co/guide/en/elasticsearch/hadoop/current/spark.html\#spark][Link 1]

ES Spark Configuration: [https://www.elastic.co/guide/en/elasticsearch/hadoop/current/configuration.html][Link 2]

end.

[https_www.cnblogs.com_upupfeng_p_12205657.html]: https://www.cnblogs.com/upupfeng/p/12205657.html
[Link 1]: https://www.elastic.co/guide/en/elasticsearch/hadoop/current/spark.html#spark
[Link 2]: https://www.elastic.co/guide/en/elasticsearch/hadoop/current/configuration.html
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2lmZW5nZ2VnZQ_size_16_color_FFFFFF_t_70]: /images/20221022/61b7457da805465cb3a64fd3f0624334.png
[20200117142726916.png]: https://img-blog.csdnimg.cn/20200117142726916.png