ClassNotFoundException：使用 elasticsearch-hadoop for Spark 时的 scala.Product$class

Question

我正在 AWS EMR 上运行 Spark-Submit 作业 从 Elasticsearch 节点读取。

当作业执行此命令时

蟒蛇：

es_config = {
    "es.nodes": url_to_my_node,
    "es.port": "9200",
    "es.resource": "my_elasticsearch_index/_doc",
    "es.query": "?q=id:park_rocky-mountain",
    "es.read.metadata": "true",
    "es.nodes.wan.only": "true",
}

df = spark.read.format("org.elasticsearch.spark.sql") \
    .options(**es_config) \
    .load()

或者在scala中

val esReadOptions = Map(
    "es.nodes" -> "url_to_my_node",
    "es.port" -> "9200"
)

val df = spark.read.format("org.elasticsearch.spark.sql")
    .options(esReadOptions)
    .load("url_to_my_node")

我收到以下错误：

: java.lang.NoClassDefFoundError: scala/Product$class
        at org.elasticsearch.spark.sql.ElasticsearchRelation.<init>(DefaultSource.scala:221)
        at org.elasticsearch.spark.sql.DefaultSource.createRelation(DefaultSource.scala:97)
        at org.apache.spark.sql.execution.datasources.DataSource.resolveRelation(DataSource.scala:350)
        at org.apache.spark.sql.DataFrameReader.loadV1Source(DataFrameReader.scala:228)
        at org.apache.spark.sql.DataFrameReader.$anonfun$load$2(DataFrameReader.scala:210)
        at scala.Option.getOrElse(Option.scala:189)
        at org.apache.spark.sql.DataFrameReader.load(DataFrameReader.scala:210)
        at org.apache.spark.sql.DataFrameReader.load(DataFrameReader.scala:171)
        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.ClientServerConnection.waitForCommands(ClientServerConnection.java:182)
        at py4j.ClientServerConnection.run(ClientServerConnection.java:106)
        at java.lang.Thread.run(Thread.java:750)
Caused by: java.lang.ClassNotFoundException: scala.Product$class
        at java.net.URLClassLoader.findClass(URLClassLoader.java:387)
        at java.lang.ClassLoader.loadClass(ClassLoader.java:418)
        at sun.misc.Launcher$AppClassLoader.loadClass(Launcher.java:352)
        at java.lang.ClassLoader.loadClass(ClassLoader.java:351)

显然这是因为找不到 JAR 或没有正确的版本。

我已经检查了 Spark 上下文中是否存在 JAR：

println(System.getProperty("java.class.path"))
> ...
> :/usr/lib/spark/jars/elasticsearch-hadoop-7.17.6.jar
> ...

关于版本冲突， EMR实例提供以下环境：

EMR 版本 3.3.2-amzn-0 使用 Scala 版本 2.12.15 OpenJDK 64 位服务器虚拟机， Java 1.8.0_382

我正在使用这个 JAR： /usr/lib/spark/jars/elasticsearch-hadoop-7.17.6.jar 它应该与 scala2.12 和 Spark 3.x 兼容

我尝试了多个不同的 JAR：

• elasticsearch-hadoop-8.9.0.jar
• elasticsearch-hadoop-7.17.6.jar
• opensearch-hadoop-1.0.1.jar

我尝试连接到不同的 Elasticsearch 实例：

• Elasticsearch 沙盒（截至撰写本文时最新）版本 8.9.0，
• Opensearch 实例（向后兼容 Elasticsearch 7.12） 我之前已经对连接进行了积极测试：我可以使用来自环境的curl 或 python.requests 通过 REST API 调用来访问这 2 个实例。

我已尝试连接

• 使用调用 pyspark 的 python 代码
• 直接从 Spark-Shell 使用 scala。 我收到与上面指定的相同的 JAVA 错误。

我已尝试连接

• 来自 AWS EMR 实例
• 来自在内核 5.15.0-78-generic 的 linux mint 20.3 上运行的本地 Spark 实现

类 scala/Product 从 2.3 版本开始就已经成为 Scala 标准库的一部分，这里我使用的是 Scala 2.12 版本。所以错误消息一定隐藏了另一个问题。

Answer 1

最终解决方案是

• 包括一个额外的 JAR
• 找到正确的配置来连接到 AWS OpenSearch

我们使用了以下内容：

AWS 开放搜索 “版本号”：“7.10.2”， “lucene_版本”：“8.10.1”

AWS EMR 与 火花版本3.3.2-amzn-0 Scala 版本 2.12.15 Java 1.8.0_382

添加了以下 JAR：

opensearch-hadoop-1.0.1.jar aws-java-sdk-1.12.531.jar

并使用以下配置：

op_config = {
    "opensearch.resource": index,
    "opensearch.nodes": elastic_host,
    "opensearch.port": elastic_port,
    "opensearch.net.ssl": "true",
    "opensearch.net.ssl.cert.allow.self.signed": "true",
    "opensearch.nodes.wan.only": "true",
    "opensearch.aws.sigv4.enabled": "true",
    "opensearch.aws.sigv4.region": aws_region,
    "opensearch.http.timeout": "3m",
}

df = spark.read.format("opensearch").options(**op_config).load()

不幸的是，github页面上的信息
https://github.com/opensearch-project/opensearch-hadoop
仅提供 Java 和 Scala 中的示例，但将其移植到 pyspark 是可行的，并且它是截至目前或撰写时发现的唯一有效文档。