@羲凡——只为了更好的活着

Spark2.0 读写ES数据(scala)——elasticsearch6.5.4

特别强调楼主使用spark2.3.2版本，elasticsearch6.5.4版本

1.准备工作

在pom.xml文件中要添加

<dependency>
	<groupId>org.elasticsearch</groupId>
	<artifactId>elasticsearch-hadoop</artifactId>
	<version>6.5.4</version>
</dependency>

特别强调如果你spark用的是2.0的版本，向es2.0的版本中插入数据要么报错要么插入失败，而且，es各大版本之间差别很大，所以强烈建议，es用最新的稳定版本。

2.不啰嗦直接上代码

代码实现：rdd或df写入到es中，且指定 es.mapping.id，再从es中读成rdd或df

package csdn

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

object SparkWriteAndReadES {

  case class StuInfo(name: String, sex: String, age: Int)

  def main(args: Array[String]): Unit = {
    val spark = SparkSession.builder()
      .appName("SparkWriteAndReadES")
      .master("local[*]")
      .config("es.index.auto.create", "true")
      .config("es.nodes", "deptest30,deptest33,deptest34")
      .config("es.port", "9200")
      .getOrCreate()

    val sc = spark.sparkContext
    val rdd = sc.makeRDD(Seq(
      StuInfo("diaochan", "女", 34),
      StuInfo("dianwei", "男", 36),
      StuInfo("guojia", "男", 35)))
    val map = Map("es.mapping.id" -> "name")

    EsSpark.saveToEs(rdd, "/stuinfo/type", map)
    println("============RDD写入ES成功！！！=================")
    val resRdd = EsSpark.esRDD(sc, "/stuinfo/type")
    println("============用esRDD读取ES结果如下：=================")
    resRdd.foreach(println)


    val df = spark.createDataFrame(sc.parallelize(Seq(
      StuInfo("xiaoming", "男", 18),
      StuInfo("xiaohong", "女", 17),
      StuInfo("xiaozhao", "男", 19)))).toDF("name", "sex", "age")

    EsSparkSQL.saveToEs(df, "/stuinfo/type", map)
    println("============RDD写入ES成功！！！=================")
    val esQuery =
      """
        |{
        |  "query": {
        |    "match": {
        |      "sex":"男"
        |    }
        |  }
        |}
      """.stripMargin
    val resDf = EsSparkSQL.esDF(spark, "/stuinfo/type",esQuery)
    println("============用esDF读取ES结果如下：=================")
    resDf.orderBy("name").show(false)

    spark.stop()
  }
}

3.结果检验

a. 检查写入是否成功,查询es结果如下
输入curl -H "Content-Type: application/json" -XGET 'deptest33:9200/stuinfo/_search?pretty' -d'{"query":{"match_all":{}}}'，出现如下结果则表示成功

{
  "took" : 10,
  "timed_out" : false,
  "_shards" : {
    "total" : 5,
    "successful" : 5,
    "skipped" : 0,
    "failed" : 0
  },
  "hits" : {
    "total" : 6,
    "max_score" : 1.0,
    "hits" : [
      {
        "_index" : "stuinfo",
        "_type" : "type",
        "_id" : "xiaozhao",
        "_score" : 1.0,
        "_source" : {
          "name" : "xiaozhao",
          "sex" : "男",
          "age" : 19
        }
      },
      {
        "_index" : "stuinfo",
        "_type" : "type",
        "_id" : "diaochan",
        "_score" : 1.0,
        "_source" : {
          "name" : "diaochan",
          "sex" : "女",
          "age" : 34
        }
      },
      {
        "_index" : "stuinfo",
        "_type" : "type",
        "_id" : "dianwei",
        "_score" : 1.0,
        "_source" : {
          "name" : "dianwei",
          "sex" : "男",
          "age" : 36
        }
      },
      {
        "_index" : "stuinfo",
        "_type" : "type",
        "_id" : "xiaoming",
        "_score" : 1.0,
        "_source" : {
          "name" : "xiaoming",
          "sex" : "男",
          "age" : 18
        }
      },
      {
        "_index" : "stuinfo",
        "_type" : "type",
        "_id" : "guojia",
        "_score" : 1.0,
        "_source" : {
          "name" : "guojia",
          "sex" : "男",
          "age" : 35
        }
      },
      {
        "_index" : "stuinfo",
        "_type" : "type",
        "_id" : "xiaohong",
        "_score" : 1.0,
        "_source" : {
          "name" : "xiaohong",
          "sex" : "女",
          "age" : 17
        }
      }
    ]
  }
}

b. 检查读取是否成功,IDEA控制台打印结果展示

============RDD写入ES成功！！！=================
============用esRDD读取ES结果如下：=================
(guojia,Map(name -> guojia, sex -> 男, age -> 35))
(diaochan,Map(name -> diaochan, sex -> 女, age -> 34))
(dianwei,Map(name -> dianwei, sex -> 男, age -> 36))
============RDD写入ES成功！！！=================
============用esDF读取ES结果如下：=================
+---+--------+---+
|age|name    |sex|
+---+--------+---+
|36 |dianwei |男  |
|34 |diaochan|女  |
|35 |guojia  |男  |
|18 |xiaoming|男  |
|19 |xiaozhao|男  |
+---+--------+---+

====================================================================
@羲凡——只为了更好的活着

若对博客中有任何问题，欢迎留言交流