如何在PySpark中将数据框保存到Elasticsearch？

Question

我有一个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

Answer 1

TL;博士 使用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

这看起来像是文档的问题（因为它们使用jar文件的两个不同版本），但这确实意味着您必须在Spark应用程序的CLASSPATH上使用正确的jar文件。

然后在同一文档中：

可以在org.elasticsearch.spark.sql软件包下获得Spark SQL支持。

这仅表示（中的df.write.format('org.elasticsearch.spark.sql')）格式正确。

在文档的更下方，您甚至可以使用别名df.write.format("es")（！）。

我在GitHub上的项目存储库中找到了Apache Spark部分，更具可读性和最新性。