我有一个Spark数据框,我试图将其推送到AWS Elasticsearch,但是在此之前,我正在测试此示例代码段以推送到ES,
from pyspark.sql import SparkSession spark = SparkSession.builder.appName('ES_indexer').getOrCreate() df = spark.createDataFrame([{'num': i} for i in xrange(10)]) df = df.drop('_id') df.write.format( 'org.elasticsearch.spark.sql' ).option( 'es.nodes', 'http://spark-data-push-adertadaltdpioy124.us-west-2.es.amazonaws.com' ).option( 'es.port', 9200 ).option( 'es.resource', '%s/%s' % ('index_name', 'doc_type_name'), ).save()
我收到一个错误消息,
java.lang.ClassNotFoundException:无法找到数据源:org.elasticsearch.spark.sql。 请在http://spark.apache.org/third-party- projects.html中找到软件包
任何建议将不胜感激。
错误跟踪:
Traceback (most recent call last): File "es_3.py", line 12, in <module> 'es.resource', '%s/%s' % ('index_name', 'doc_type_name'), File "/usr/local/lib/python2.7/site-packages/pyspark/sql/readwriter.py", line 732, in save self._jwrite.save() File "/usr/local/lib/python2.7/site-packages/py4j/java_gateway.py", line 1257, in __call__ answer, self.gateway_client, self.target_id, self.name) File "/usr/local/lib/python2.7/site-packages/pyspark/sql/utils.py", line 63, in deco return f(*a, **kw) File "/usr/local/lib/python2.7/site-packages/py4j/protocol.py", line 328, in get_return_value format(target_id, ".", name), value) py4j.protocol.Py4JJavaError: An error occurred while calling o46.save. : java.lang.ClassNotFoundException: Failed to find data source: org.elasticsearch.spark.sql. Please find packages at http://spark.apache.org/third-party-projects.html at org.apache.spark.sql.execution.datasources.DataSource$.lookupDataSource(DataSource.scala:657) at org.apache.spark.sql.DataFrameWriter.save(DataFrameWriter.scala:245) at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method) at sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62) at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43) at java.lang.reflect.Method.invoke(Method.java:498) at py4j.reflection.MethodInvoker.invoke(MethodInvoker.java:244) at py4j.reflection.ReflectionEngine.invoke(ReflectionEngine.java:357) at py4j.Gateway.invoke(Gateway.java:282) at py4j.commands.AbstractCommand.invokeMethod(AbstractCommand.java:132) at py4j.commands.CallCommand.execute(CallCommand.java:79) at py4j.GatewayConnection.run(GatewayConnection.java:238) at java.lang.Thread.run(Thread.java:748) Caused by: java.lang.ClassNotFoundException: org.elasticsearch.spark.sql.DefaultSource at java.net.URLClassLoader.findClass(URLClassLoader.java:382) at java.lang.ClassLoader.loadClass(ClassLoader.java:424) at java.lang.ClassLoader.loadClass(ClassLoader.java:357) at org.apache.spark.sql.execution.datasources.DataSource$$anonfun$20$$anonfun$apply$12.apply(DataSource.scala:634) at org.apache.spark.sql.execution.datasources.DataSource$$anonfun$20$$anonfun$apply$12.apply(DataSource.scala:634) at scala.util.Try$.apply(Try.scala:192) at org.apache.spark.sql.execution.datasources.DataSource$$anonfun$20.apply(DataSource.scala:634) at org.apache.spark.sql.execution.datasources.DataSource$$anonfun$20.apply(DataSource.scala:634) at scala.util.Try.orElse(Try.scala:84) at org.apache.spark.sql.execution.datasources.DataSource$.lookupDataSource(DataSource.scala:634) ... 12 more
TL;博士 使用pyspark --packages org.elasticsearch:elasticsearch- hadoop:7.2.0和使用format("es"),以引用连接器。
pyspark --packages org.elasticsearch:elasticsearch- hadoop:7.2.0
format("es")
从Elasticsearch for Apache Hadoop产品的官方文档中引用安装:
就像其他库一样,elasticsearch-hadoop必须在Spark的类路径中可用。
以及稍后在受支持的Spark SQL版本中:
elasticsearch-hadoop通过两个不同的jar支持Spark SQL 1.3-1.6版本和Spark SQL 2.0版本:elasticsearch-spark-1.x-<version>.jar和elasticsearch- hadoop-<version>.jar elasticsearch-spark-2.0-<version>.jar 支持Spark SQL 2.0
elasticsearch-hadoop通过两个不同的jar支持Spark SQL 1.3-1.6版本和Spark SQL 2.0版本:elasticsearch-spark-1.x-<version>.jar和elasticsearch- hadoop-<version>.jar
elasticsearch-spark-1.x-<version>.jar
elasticsearch- hadoop-<version>.jar
elasticsearch-spark-2.0-<version>.jar 支持Spark SQL 2.0
elasticsearch-spark-2.0-<version>.jar
这看起来像是文档的问题(因为它们使用jar文件的两个不同版本),但这确实意味着您必须在Spark应用程序的CLASSPATH上使用正确的jar文件。
然后在同一文档中:
可以在org.elasticsearch.spark.sql软件包下获得Spark SQL支持。
这仅表示(中的df.write.format('org.elasticsearch.spark.sql'))格式正确。
df.write.format('org.elasticsearch.spark.sql')
在文档的更下方,您甚至可以使用别名df.write.format("es")(!)。
df.write.format("es")
我在GitHub上的项目存储库中找到了Apache Spark部分,更具可读性和最新性。