sparkSql 读写Es

你的名字 2022-09-03 03:24 261阅读 0赞

**目录**

一、maven相关依赖

二、ticket.csv

三、代码

四、输出结果

五、Es版本问题

--------------------

# 一、maven相关依赖 #

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

# 二、ticket.csv #

id,date,openPrice,highestPrice,lowestPrice,closePrice,exchangeNum
    SH600000,2015-7-1 9:31:00,137.5999,137.5999,136.5358,136.7814,64912
    SH600000,2015-7-1 9:32:00,136.8632,137.5181,136.6995,137.5181,31363
    SH600000,2015-7-1 9:33:00,137.4362,137.8455,137.1088,137.6818,25057
    SH600000,2015-7-1 9:34:00,137.5999,137.5999,136.7814,136.8632,32136
    SH600000,2015-7-1 9:35:00,137.1088,137.3544,136.8632,137.0269,29211
    ...

# 三、代码 #

package com.dt.spark.Test
    
    import org.apache.spark.{SparkConf, SparkContext}
    import org.apache.spark.sql.{DataFrame, SparkSession}
    import org.elasticsearch.spark.sql.EsSparkSQL
    object SaveToEs {
      def main(args: Array[String]): Unit = {
        val sparkConf = new SparkConf().setAppName("saveToEsTest")
          .setMaster("local")
        sparkConf.set("es.nodes","192.168.xx.xx")
        sparkConf.set("es.port","9200")
        sparkConf.set("es.index.auto.create", "true")
        sparkConf.set("es.write.operation", "index")
        sparkConf.set("es.nodes.wan.only","true")
        val spark: SparkSession = SparkSession.builder().config(sparkConf).getOrCreate()
        val sc: SparkContext = spark.sparkContext
        sc.setLogLevel("WARN")
    
        val df: DataFrame = spark.read
          //推断数据类型
          .option("inferSchema", "true")
          //表示有表头，若没有则为false
          .option("header", true)
          .csv("data/test/ticket.csv")
        //将数据写入ES
        EsSparkSQL.saveToEs(df,"test/myType")  //索引/类型
    
        //从Es读取数据
        spark.read.format("es").load("test/myType").show()
    
      }
    }

# 四、输出结果 #

+----------+-------------+-----------+------------+--------+-----------+---------+
    |closePrice|         date|exchangeNum|highestPrice|      id|lowestPrice|openPrice|
    +----------+-------------+-----------+------------+--------+-----------+---------+
    |  137.0269|1435714500000|      29211|    137.3544|SH600000|   136.8632| 137.1088|
    |  137.1088|1435714800000|      25467|    138.0092|SH600000|   136.7814| 136.9451|
    |  137.1907|1435714860000|      41187|    137.5999|SH600000|   136.8632| 137.1088|
    |  137.1088|1435715460000|       5221|    137.5181|SH600000|   137.1088| 137.5181|
    |  137.6818|1435715700000|      10690|    137.6818|SH600000|   137.1088| 137.1088|
    |  137.5999|1435715760000|      17265|    137.6818|SH600000|   137.5181| 137.5181|
    |  137.3544|1435715880000|      12858|    137.8455|SH600000|   137.3544| 137.6818|
    |  136.7814|1435716540000|      10546|    136.9451|SH600000|   136.7814| 136.8632|
    |  136.2084|1435717140000|      15679|    136.5358|SH600000|   136.0447| 136.3721|
    |  136.6995|1435717500000|       6604|    136.8632|SH600000|   136.6177| 136.6995|
    |  136.9451|1435717620000|       5395|    137.1907|SH600000|   136.7814| 137.1088|

# 五、Es版本问题 #

ES 7.x以后已经移除type这个概念,默认的type为\_doc,故有7.x以上有两种读取和存储方法:

方法一:
     //将数据写入ES
     EsSparkSQL.saveToEs(df,"test/_doc")  //索引/类型
    
     //从Es读取数据
     spark.read.format("es").load("test/_doc")
    
    方法二:
     //将数据写入ES
     EsSparkSQL.saveToEs(df,"test")  //索引
    
     //从Es读取数据
     spark.read.format("es").load("test")