1. spark 读取 ES

import org.apache.spark.sql.SparkSession
import org.elasticsearch.spark.rdd.EsSpark

object esReadToHdfs {
  def main(args: Array[String]): Unit = {
    val spark = SparkSession.builder().appName("es_read").getOrCreate()
    val sc = spark.sparkContext

    val options = Map(
      "es.index.auto.create" -> "true",
      "es.nodes.wan.only" -> "true",
      "es.nodes" -> "29.29.29.29:9200,29.29.29.29:9200",
      "es.port" -> "9200",
      "es.mapping.id" -> "id"
    )

    // 返回 RDD[(String, String]]
    // 元组：第一个：esmapping.id、第二个 json 字符串
    val resultRDD = EsSpark.esJsonRDD(sc, options).map(x => x._2)

    //    // 返回 RDD[(String, Map[String, AnyDef]]
    //    val resultRDD = EsSpark.esRDD(sc, options)

  }
}

2. spark 写入 ES

参考文章

• Spark读写ES数据时遇到的问题总结
• Spark读写ES