从 Spark Streaming Job 写入多个 ES sink

Question

我正在使用 Spark 流处理来自 Kafka 队列的大量数据，我需要将结果写入两个已设置的 ES 集群。

我在我的项目中使用“org.elasticsearch.elasticsearch-spark-XX”，并且能够通过使用方法 EsSpark.saveJsonToEs() 写入一个 ES 集群，该方法在内部从不可变的 spark 上下文中获取 ES 特定属性。

所以，如果 spark 上下文是不可变的，我如何在运行时更改属性并将结果写入多个 ES 集群。

请提出建议。

参考 - https://www.elastic.co/guide/en/elasticsearch/hadoop/6.7/spark.html

Answer 1

恕我直言，spark.driver.allowMultipleContexts 的以下方式应该可以工作...... 使用新的 spark 配置分别创建 spark 上下文会话。

conf.set("es.index.auto.create", "true")
conf.set("es.nodes", "XXXXXXXX")
conf.set("es.port", "9020")
conf.set("spark.driver.allowMultipleContexts", "true")
val sc1 = new SparkContext(conf)

---

conf.set("es.index.auto.create", "true")
conf.set("es.nodes", "yyyyyy")
conf.set("es.port", "9020")
conf.set("spark.driver.allowMultipleContexts", "true")
val sc2 = new SparkContext(conf)