Spark流数据帧对elasticsearch公开

Question

SparkSession配置：

val spark = SparkSession.builder()
    .config("es.index.auto.create", "true")
    .config("es.nodes", "1.2.3.4")
    .config("es.port", "9200")
    .config("es.nodes.wan.only", "true")
    .master("local[*]")
    .getOrCreate()

保存查询：

val query = tableDF
    .writeStream
    .format("es")
    .outputMode("append")
    .option("checkpointLocation", "/tmp/checkpoint")
    .option("resource", "spark/speeches")
    .start()

query.awaitTermination()

错误

Caused by: java.lang.NoSuchMethodError: org.apache.spark.sql.execution.SQLExecution$.withNewExecutionId(Lorg/apache/spark/sql/SparkSession;Lorg/apache/spark/sql/execution/QueryExecution;Lscala/Function0;)Ljava/lang/Object;

我的依赖项：

compile group: 'org.apache.spark', name: 'spark-sql_2.12', version: '3.0.1'
compile group: 'org.apache.spark', name: 'spark-core_2.12', version: '3.0.1'
compile group: 'org.apache.spark', name: 'spark-streaming_2.12', version: '3.0.1'
implementation group: 'org.elasticsearch', name: 'elasticsearch-hadoop', version: '7.11.1'

Answer 1

替换依赖关系

implementation group: 'org.elasticsearch', name: 'elasticsearch-hadoop', version: '7.11.1'

至

implementation group: 'org.elasticsearch', name: 'elasticsearch-spark-20_2.11', version: '7.11.1'

您应该更改使用scala 2.11的其他依赖项