使用ES-Hadoop插件结合spark向es插入数据

墨蓝 2022-06-16 04:45 177阅读 0赞

上篇文章简单介绍了ES-Hadoop插件的功能和使用场景，本篇就来看下如何使用ES-Hadoop里面的ES-Spark插件，来完成使用spark想es里面大批量插入数据。

这里说明下ES-Hadoop是一个fat的maven依赖，也就是引入这个包会导入许多额外的依赖，不建议这么搞，除非是测试为了方便用的，实际情况是ES也单独提供了每个不同框架的mini包，比如说只使用spark那么就可以单独引入es-spark的包，而不需要使用fat包因为它会附加引入其他的包依赖，如果使用hive那就单独引入es-hive的包即可。这里面唯一需要注意的就是注意版本问题，不同的框架的版本和es-hadoop的组件都有对应的版本，使用不当会报异常，具体的情况，请参考官网文档：

[https://www.elastic.co/guide/en/elasticsearch/hadoop/current/install.html][https_www.elastic.co_guide_en_elasticsearch_hadoop_current_install.html]

下面看下如何使用es-spark读写es的数据：

spark版本：2.1.0

scala版本：2.11.8

es版本：2.3.4

索引要引入下面的依赖pom才行

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

这里为了快速体验，所以直接使用spark的local模式测试，如果要放到正式环境运行，切记把local模式的代码去掉。

先看下向es里面插入数据的代码，非常简单：

val conf=new SparkConf()
        conf.setMaster("local[1]")//指定local模式
        conf.setAppName("spark to es")//设置任务名
        conf.set("es.index.auto.create","true")//开启自动创建索引
        conf.set("es.nodes","192.168.201.5,192.168.201.6")//es的节点，多个用逗号分隔
        conf.set("es.port","9200")//端口号
        val sc=new SparkContext(conf)
        val data1 = Map("id" -> 1, "name" -> "tom", "age" -> 19)//第一条数据
        val data2 = Map("id" -> 2, "name" -> "john","age"->25)//第二条数据
        sc.makeRDD(Seq(data1, data2)).saveToEs("spark/docs")//添加到索引里面
        println("存储成功！")
        sc.stop()

然后我们看下如何读取es里面的数据：

val conf=new SparkConf()
        conf.setMaster("local[1]")//指定local模式
        conf.setAppName("spark to es")//任务名
        conf.set("es.nodes","192.168.201.5,192.168.201.6")//es节点多个逗号分隔
        conf.set("es.port","9200")
        val sc=new SparkContext(conf)
        val ds=sc.esRDD("stu2017-05-03/stu")//读取数据到spark的rdd里面
        log.info("stu2017-05-03数据总量："+ds.count())//统计数量
        sc.stop()

从上面的代码量我们可以看到非常少，这是由于es-spark底层已经帮我们封装好了相关的代码，所以用起来非常简单，围绕的核心还是rdd，无论是写入es，还是从es读取数据都是通过spark的rdd做中转的，我们只要把我们的目标数据给转成RDD或者DataFrame就能非常方便的与es对接了。

上面的代码使用spark的core来完成的，此外我门还可以使用spark sql或者spark streaming来与es对接，这个以后用到的时候再总结分享，最后使用spark操作es的时候我门还可以有非常多的配置参数设置，本例子中只使用了部分的参数，关于详细的配置参数 大家可以看下面的这个类：

org.elasticsearch.hadoop.cfg.ConfigurationOptions

官网文档：

[https://www.elastic.co/guide/en/elasticsearch/hadoop/current/reference.html][https_www.elastic.co_guide_en_elasticsearch_hadoop_current_reference.html]

有什么问题可以扫码关注微信公众号：我是攻城师(woshigcs)，在后台留言咨询。 技术债不能欠，健康债更不能欠， 求道之路，与君同行。

![输入图片说明][3214000f-5633-3c17-a3d7-83ebda9aebff.jpg]

[https_www.elastic.co_guide_en_elasticsearch_hadoop_current_install.html]: https://www.elastic.co/guide/en/elasticsearch/hadoop/current/install.html
[https_www.elastic.co_guide_en_elasticsearch_hadoop_current_reference.html]: https://www.elastic.co/guide/en/elasticsearch/hadoop/current/reference.html
[3214000f-5633-3c17-a3d7-83ebda9aebff.jpg]: /images/20220616/7aa3e399974441afbeb138951f74a627.png