Spark讀寫ES

本文轉載自查看原文 2020-01-17 14:36 10738 ElasticSearch/ Spark ES

本文主要介紹spark sql讀寫es、structured streaming寫入es以及一些參數的配置

ES官方提供了對spark的支持，可以直接通過spark讀寫es，具體可以參考ES Spark Support文檔(文末有地址)。

以下是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實現。在實現前，還要列一些相關的配置：

配置

參數	描述
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：資源路徑，例如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：資源路徑，例如hive_table/docs

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

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.table("hive_table")
df.saveToEs("hive_table/docs", 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()