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

Question

我有一个 spark 数据框，我正在尝试推送到 AWS Elasticsearch，但在此之前我正在测试这个示例 codesn-p 以推送到 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;dr 使用 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 通过两个不同的 jars 支持 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 文件。

后来在同一个document:

org.elasticsearch.spark.sql 包下提供 Spark SQL 支持。

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

在document 的下方，您会发现您甚至可以使用别名df.write.format("es") (!)

我发现 GitHub 上项目存储库中的 Apache Spark 部分更具可读性和最新性。

Answer 2

更新：截至 2020 年 6 月，当前的 ES-hadoop 包是 7.7.1，所以我改用了pyspark --packages org.elasticsearch:elasticsearch-hadoop:7.7.1。