apache-spark - Spark 流 : Write dataframe to ElasticSearch

Question

我使用以下代码从 python (pyspark) 应用程序向 elasticsearch 写入一个流。

#Streaming code
query = df.writeStream \
.outputMode("append") \
.format("org.elasticsearch.spark.sql") \
.option("checkpointLocation", "/tmp/") \
.option("es.resource", "logs/raw") \
.option("es.nodes", "localhost") \
.start()

query.awaitTermination()

如果我将结果写入控制台它工作正常，同样，如果我写入 ES - 不是在流模式下，它工作正常。这是我以前写到 ES 的代码:

#Not streaming
df.write.format("org.elasticsearch.spark.sql") \
.mode('append') \
.option("es.resource", "log/raw") \
.option("es.nodes", "localhost").save("log/raw")

问题是，我无法调试它，代码正在运行，但没有任何内容写入 ES(在流模式下)。

谢谢，

Answer 1

最终确实为我解决了，问题是技术性的(需要 vpn)

query = df.writeStream \
.outputMode("append") \
.queryName("writing_to_es") \
.format("org.elasticsearch.spark.sql") \
.option("checkpointLocation", "/tmp/") \
.option("es.resource", "index/type") \
.option("es.nodes", "localhost") \
.start()

query.awaitTermination()