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

Question

我有一个spark dataframe，我想将其推送到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

- Cyber_Tron

Answer 1

使用 pyspark --packages org.elasticsearch:elasticsearch-hadoop:7.2.0 命令，并使用 format("es") 引用连接器。

---

引用Elasticsearch for Apache Hadoop产品的官方文档Installation中的内容：

与其他库一样，elasticsearch-hadoop需要在Spark的类路径中可用。

稍后在Supported Spark SQL versions中提到：

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文件。

在同一文档中稍后:

Spark SQL支持可在org.elasticsearch.spark.sql包下使用。

这仅仅是说格式（在df.write.format('org.elasticsearch.spark.sql')）是正确的。

在文档更深处，您甚至可以发现您可以使用别名df.write.format("es")！

我在GitHub上项目的Apache Spark部分更易读且更为最新。

- Jacek Laskowski

Answer 2

更新：截至2020年6月，当前的ES-hadoop包版本为7.7.1，因此我使用了pyspark --packages org.elasticsearch:elasticsearch-hadoop:7.7.1。

- Ezra Justin Lee

Answer 3

在包的末尾必须提及您的elasticsearch数据库版本，例如--packages org.elasticsearch:elasticsearch-hadoop:(version)。在我的情况下，它是org.elasticsearch:elasticsearch-hadoop:7.0.0。

- Hamza Younis