【spark工具库】spark sql读写es操作

Spark SQL可以通过Elasticsearch-Hadoop库来读写Elasticsearch数据。该库提供了一个Elasticsearch数据源，可以通过Spark SQL的DataFrame API来读写Elasticsearch数据。要使用该库，需要在Spark应用程序中添加以下依赖项：

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

然后，可以使用以下代码来读取Elasticsearch数据：

val df = spark.read.format("org.elasticsearch.spark.sql")
    .option("es.nodes", "localhost")
    .option("es.port", "9200")
    .option("es.index.auto.create", "true")
    .load("index_name/type_name")

其中，es.nodes和es.port指定了Elasticsearch的主机和端口，es.index.auto.create指定了是否自动创建索引，load方法指定了要读取的索引和类型。

要将数据写入Elasticsearch，可以使用以下代码：

df.write.format("org.elasticsearch.spark.sql")
    .option("es.nodes", "localhost")
    .option("es.port", "9200")
    .option("es.index.auto.create", "true")
    .mode("overwrite")
    .save("index_name/type_name")

其中，mode指定了写入模式，可以是overwrite、append或ignore。save方法指定了要写入的索引和类型。