问题描述
我正在编写一个代码，其中我试图使用pySpark的结构化流将数据流化为弹性搜索。
火花版本：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（本机方法）处 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个
感谢?的帮助！
解决方法
非常感谢，我使用的是基于Scala 2.12构建的spark 3，不幸的是，直到2.11版本的scala都支持elasticsearch-hadoop jar。我已将我的Spark版本降级为基于Scala 2.11构建的2.4.6。