将数据从Apache spark中的JavaDStream<String>写入到elasticsearch

<dependencies>
    <dependency>
        <groupId>org.elasticsearch</groupId>
        <artifactId>elasticsearch-spark-20_2.11</artifactId>
        <version>7.15.0</version>
    </dependency>
    <dependency>
        <groupId>org.apache.spark</groupId>
        <artifactId>spark-core_2.11</artifactId>
        <version>2.4.8</version>
    </dependency>
    <dependency>
        <groupId>org.apache.spark</groupId>
        <artifactId>spark-streaming_2.11</artifactId>
        <version>2.4.8</version>
    </dependency>
</dependencies>

import org.apache.spark.SparkConf;
import org.apache.spark.streaming.api.java.JavaDStream;
import org.apache.spark.streaming.api.java.JavaStreamingContext;

SparkConf conf = new SparkConf()
    .setAppName("Write to Elasticsearch")
    .setMaster("local[*]"); // 设置Spark Master URL

JavaStreamingContext jssc = new JavaStreamingContext(conf, Durations.seconds(1));

JavaDStream<String> dataStream = jssc.socketTextStream("localhost", 9999);

import org.elasticsearch.spark.rdd.api.java.JavaEsSpark;

Map<String, String> esConfig = new HashMap<>();
esConfig.put("es.nodes", "localhost");
esConfig.put("es.port", "9200");

dataStream.foreachRDD(rdd -> {
    JavaEsSpark.saveJsonToEs(rdd, "index_name/type_name", esConfig);
});

jssc.start();
jssc.awaitTermination();