搭建参考:我的这篇
elasticsearch搭建_我要用代码向我喜欢的女孩表白的博客-CSDN博客
为了方便测试，我们先建立个索引，如果没有索引，他也能插入，只是走的是默认插入格式。
不过虽然接触es已经4年了，但是在工作中，整体项目用es存储的机会还是比较少，所以我对索引不太熟悉，这里，就请大家勉为其难和我一起学习下，es的索引（熟手可跳过，直接看下面的开发部分）
下面需要记住的话
1.索引就相当于是es的数据库（7.x版本的es没有数据表的概念）
PUT http://IP:9200/test1
这个api能够建索引，默认情况下，创建的索引分片数量是 5 个，副本数量是 1 个。我的是单节点测试，所以分片和副本都不用管。分片的意思是将数据分成多少个部分，副本意思是多复制几份。如果是2个副本，那相当于总共创建了10个分片。
{
"settings": {
"number_of_shards": 2,
"number_of_replicas": 1
}
}
shards代表分片，repicas代表副本
修改索引，test1/_settings（意思为:索引名/设置索引)
刚刚看了尚硅谷的es索引那块，太基础了，全使用的默认的，具体mapping分片之类的，如何分词都没有讲。
如果删除索引，请求类型就是DELTE
ip:9200/索引名
get类型查看索引
ip:9200/_cat/indices?v
显示所有索引
版本选择
我的ES版本是7.9.3
spark是2.4
scala是2.11.11
这几个是互相兼容的
Spark连接Es并且插入数据
pom导入依赖(scala版本)
注:java版本的请看参考连接中的阿里云ElasticSearch
elasticsearch-hadoop 和 elasticsearch-spark 是不同的两个连接器，es-hadoop是hive的，es-spark是spark的。并且不能互相存在，会提示冲突。建表语法也不一样。这里我们不用hql，纯用spark所以导入es-spark
<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0"
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
<modelVersion>4.0.0</modelVersion>
<groupId>spark</groupId>
<artifactId>sparkEs</artifactId>
<version>1.0-SNAPSHOT</version>
<!-- https://mvnrepository.com/artifact/org.elasticsearch/elasticsearch-spark-20 -->
<dependencies>
<dependency>
<groupId>org.apache.spark</groupId>
<artifactId>spark-core_2.11</artifactId>
<version>2.4.5</version>
</dependency>
<dependency>
<groupId>org.apache.spark</groupId>
<artifactId>spark-sql_2.11</artifactId>
<version>2.4.5</version>
</dependency>
<dependency>
<groupId>org.elasticsearch</groupId>
<artifactId>elasticsearch-spark-20_2.11</artifactId>
<version>7.9.3</version>
</dependency>
</dependencies>
</project>
代码
import org.apache.spark.{SparkConf, SparkContext}
import org.apache.spark.SparkContext
import org.elasticsearch.spark.rdd.EsSpark
object SparkEs {
def main(args: Array[String]): Unit = {
// val conf = new SparkConf().setAppName("MyScalaWordCount").setMaster("local[1]")
val conf = new SparkConf().setAppName("readEs").setMaster("local[2]")
.set("es.nodes", "es-cn-n6w1o1x0w001c****.Elasticsearch.aliyuncs.com")
.set("es.port", "9200")
.set("es.net.http.auth.user", "elastic")
.set("es.net.http.auth.pass", "xxxxxx")
.set("es.nodes.wan.only", "true")
.set("es.nodes.discovery","false")
.set("es.input.use.sliced.partitions","false")
.set("es.resource", "index_name/_doc")
.set("es.scroll.size","500");
conf.set("es.index.auto.create","true")
val sc = new SparkContext(conf)
val num = Map("a"->1,"b"->2)
EsSpark.esRDD(sc)//传入spark连接es的配置，通过ES类解析spark中es的配置，获取Es数据
val rdd = sc.makeRDD(Seq(num))
EsSpark.saveToEs(rdd,"spark/doc")
}
}参考：
spark 使用elasticsearch-spark connector读取ES 跳坑记录 - 简书
https://mvnrepository.com/artifact/org.elasticsearch/elasticsearch-spark-20_2.11/7.9.3
如何使用DLASpark访问阿里云Elasticsearch_云原生数据湖分析-阿里云帮助中心
Apache Spark support | Elasticsearch for Apache Hadoop [7.9] | Elastic
