pyspark - 结构化流到弹性搜索答案

【问题标题】：pyspark - structured streaming into elastic searchpyspark - 结构化流到弹性搜索
【发布时间】：2020-12-12 10:58:54
【问题描述】：

我正在编写一个代码，我正在尝试使用 pySpark 的结构化流将数据流式传输到弹性搜索中。

Spark 版本：3.0.0 安装模式：pip

query = inpJoinDF.writeStream \
.outputMode("append") \
.queryName("writing_to_es") \
.format("org.elasticsearch.spark.sql") \
.option("checkpointLocation", "es_checkpoint/") \
.option("es.resource", "spark_test/doc") \
.option("es.nodes", "localhost") \
.start()

也尝试过添加包和格式

pyspark --packages org.elasticsearch:elasticsearch-hadoop:7.7.1
format("es")

以下是错误行：

第 328 行，在 get_return_value py4j.protocol.Py4JJavaError: 一个错误调用 o57.start 时发生。：java.lang.ClassNotFoundException：找不到数据源：org.elasticsearch.spark.sql。请找到包裹在http://spark.apache.org/third-party-projects.html 在 org.apache.spark.sql.execution.datasources.DataSource$.lookupDataSource(DataSource.scala:674) 在 org.apache.spark.sql.streaming.DataStreamWriter.start(DataStreamWriter.scala:342) 在 sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method) 在 sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62) 在 sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43) 在 java.lang.reflect.Method.invoke(Method.java:498) 在 py4j.reflection.MethodInvoker.invoke(MethodInvoker.java:244) 在 py4j.reflection.ReflectionEngine.invoke(ReflectionEngine.java:357) 在 py4j.Gateway.invoke(Gateway.java:282) 在 py4j.commands.AbstractCommand.invokeMethod(AbstractCommand.java:132) 在 py4j.commands.CallCommand.execute(CallCommand.java:79) 在 py4j.GatewayConnection.run(GatewayConnection.java:238) 在 java.lang.Thread.run(Thread.java:748) 原因： java.lang.ClassNotFoundException： org.elasticsearch.spark.sql.DefaultSource 在 java.net.URLClassLoader.findClass(URLClassLoader.java:382) 在 java.lang.ClassLoader.loadClass(ClassLoader.java:424) 在 java.lang.ClassLoader.loadClass(ClassLoader.java:357) 在 org.apache.spark.sql.execution.datasources.DataSource$.$anonfun$lookupDataSource$5(DataSource.scala:648) 在 scala.util.Try$.apply(Try.scala:213) 在 org.apache.spark.sql.execution.datasources.DataSource$.$anonfun$lookupDataSource$4(DataSource.scala:648) 在 scala.util.Failure.orElse(Try.scala:224) 在 org.apache.spark.sql.execution.datasources.DataSource$.lookupDataSource(DataSource.scala:648) ... 12 更多

谢谢????帮忙！！

【问题讨论】：

当我使用软件包或 jar 文件时，我收到的功能未找到。使用的 jar：elasticsearch-hadoop-7.9.0.jar

标签： apache-spark elasticsearch pyspark spark-streaming spark-structured-streaming

【解决方案1】：

非常感谢，我使用的是基于 scala 2.12 构建的 spark 3，不幸的是，直到 scala 2.11 版本才支持 elasticsearch-hadoop jar。我已将我的 spark 版本降级为基于 scala 2.11 的 2.4.6。

【讨论】：