使用Apache Spark将Oracle DataFrame写入Elasticsearch遇类缺失错误求助

环境信息 #

• Elasticsearch 8.8.2
• Spark 3.4.3
• Python 3.8
• PySpark 3.4.1
• Scala 2.12

相关代码 #

write方法 #

def write(self, dataFrame):
    """start indexing new dataSource to Elasticsearch with all required steps

    Arguments:
        dataFrame {Spark DataFrame} -- source dataFrame

    Raises:
        Error -- may raise unexpected types of errors
    """
    # create new index name for this dataFrame
    self.currentIndex = self.generateIndex()
    # set Source document details before indexing
    try:
        count = dataFrame.count()
    except Exception as err:
        self.terminate("Failed to dataframe count", error=err)
    self.__updateSource(key=Source.rowNumber, value=count)

    # inform application about importing percent
    thread = Thread(target=self.__pubPercent)
    thread.start()
    # emit number of rows
    rows = self.document.get(Source.rowNumber)
    # index this dataFrame into Elasticsearch
    partitions = 10 if rows < 10**5 else 100
    try:
        self.setDataFrame(self.currentIndex, dataFrame.repartition(partitions))
    except Exception as err:
        print(err)
        # wait thread till finish
        thread.join()

setDataFrame方法 #

def setDataFrame(self, index, dataFrame, idField=None):
    """Create new index using dataFrame
    Spark doesn't accept to add dataFrame documents in existing index
    So, it'll delete index if exists

    Arguments:
        index {str} -- Index Name
        dataFrame {Spark dateFrame} -- index documents

    Keyword Arguments:
        idField {str} -- Name of identity field (default:None)

    Raises:
        Error -- may raise unexpected types of errors
    """
    # delete old index if exists
    esClient.deleteIndex(index)

    # trying to set index mapping
    esClient.putMapping(index, dataFrame.dtypes)

    # write data frame to elasticsearch

    writer = (
        dataFrame.write.format(self.SPARK_SQL_FORMAT)
        .option(self.RESOURSE_KEY, "{}/{}".format(index, index))
        .option(self.NODE_KEY, self.__host)
        .option(self.PORT_KEY, self.__port)
    )
    # check if there's identity field or not

    if idField:
        writer = writer.option(self.MAPPING_ID_KEY, idField)

    # export dataFrame into Elastic
    writer.save()

异常日志 #

An error occurred while calling o68.save.
: java.lang.NoClassDefFoundError: scala/Product$class
        at org.elasticsearch.spark.sql.ElasticsearchRelation.<init>(DefaultSource.scala:228)
        at org.elasticsearch.spark.sql.DefaultSource.createRelation(DefaultSource.scala:105)
        at org.apache.spark.sql.execution.datasources.SaveIntoDataSourceCommand.run(SaveIntoDataSourceCommand.scala:47)
        at org.apache.spark.sql.execution.command.ExecutedCommandExec.sideEffectResult$lzycompute(commands.scala:75)
        at org.apache.spark.sql.execution.command.ExecutedCommandExec.sideEffectResult(commands.scala:73)
        at org.apache.spark.sql.execution.command.ExecutedCommandExec.executeCollect(commands.scala:84)
        at org.apache.spark.sql.execution.QueryExecution$$anonfun$eagerlyExecuteCommands$1.$anonfun$applyOrElse$1(QueryExecution.scala:98)
        at org.apache.spark.sql.execution.SQLExecution$.$anonfun$withNewExecutionId$6(SQLExecution.scala:118)
        at org.apache.spark.sql.execution.SQLExecution$.withSQLConfPropagated(SQLExecution.scala:195)
        at org.apache.spark.sql.execution.SQLExecution$.$anonfun$withNewExecutionId$1(SQLExecution.scala:103)
        at org.apache.spark.sql.SparkSession.withActive(SparkSession.scala:827)
        at org.apache.spark.sql.execution.SQLExecution$.withNewExecutionId(SQLExecution.scala:65)
        at org.apache.spark.sql.execution.QueryExecution$$anonfun$eagerlyExecuteCommands$1.applyOrElse(QueryExecution.scala:98)
        at org.apache.spark.sql.execution.QueryExecution$$anonfun$eagerlyExecuteCommands$1.applyOrElse(QueryExecution.scala:94)
        at org.apache.spark.sql.catalyst.trees.TreeNode.$anonfun$transformDownWithPruning$1(TreeNode.scala:512)
        at org.apache.spark.sql.catalyst.trees.CurrentOrigin$.withOrigin(TreeNode.scala:104)
        at org.apache.spark.sql.catalyst.trees.TreeNode.transformDownWithPruning(TreeNode.scala:512)
        at org.apache.spark.sql.catalyst.plans.logical.LogicalPlan.org$apache$spark$sql$catalyst$plans$logical$AnalysisHelper$$super$transformDownWithPruning(LogicalPlan.scala:31)
        at org.apache.spark.sql.catalyst.plans.logical.AnalysisHelper.transformDownWithPruning(AnalysisHelper.scala:267)
        at org.apache.spark.sql.catalyst.plans.logical.AnalysisHelper.transformDownWithPruning$(AnalysisHelper.scala:263)
        at org.apache.spark.sql.catalyst.plans.logical.LogicalPlan.transformDownWithPruning(LogicalPlan.scala:31)
        at org.apache.spark.sql.catalyst.plans.logical.LogicalPlan.transformDownWithPruning(LogicalPlan.scala:31)
        at org.apache.spark.sql.catalyst.trees.TreeNode.transformDown(TreeNode.scala:488)
        at org.apache.spark.sql.execution.QueryExecution.eagerlyExecuteCommands(QueryExecution.scala:94)
        at org.apache.spark.sql.execution.QueryExecution.commandExecuted$lzycompute(QueryExecution.scala:81)
        at org.apache.spark.sql.execution.QueryExecution.commandExecuted(QueryExecution.scala:79)
        at org.apache.spark.sql.execution.QueryExecution.assertCommandExecuted(QueryExecution.scala:133)
        at org.apache.spark.sql.DataFrameWriter.runCommand(DataFrameWriter.scala:856)
        at org.apache.spark.sql.DataFrameWriter.saveToV1Source(DataFrameWriter.scala:387)
        at org.apache.spark.sql.DataFrameWriter.saveInternal(DataFrameWriter.scala:360)
        at org.apache.spark.sql.DataFrameWriter.save(DataFrameWriter.scala:247)
        at java.base/jdk.internal.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
        at java.base/jdk.internal.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:77)
        at java.base/jdk.internal.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
        at java.base/java.lang.reflect.Method.invoke(Method.java:568)
        at py4j.reflection.MethodInvoker.invoke(MethodInvoker.java:244)
        at py4j.reflection.ReflectionEngine.invoke(ReflectionEngine.java:374)
        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.base/java.lang.Thread.run(Thread.java:833)
Caused by: java.lang.ClassNotFoundException: scala.Product$class
        at java.base/jdk.internal.loader.BuiltinClassLoader.loadClass(BuiltinClassLoader.java:641)
        at java.base/jdk.internal.loader.ClassLoaders$AppClassLoader.loadClass(ClassLoaders.java:188)
        at java.base/java.lang.ClassLoader.loadClass(ClassLoader.java:520)

这个错误的核心原因是Elasticsearch Spark Connector依赖的Scala版本与当前Spark使用的Scala版本不兼容。Spark 3.4.x基于Scala 2.12编译，必须匹配对应Scala版本的connector包。

具体修复步骤： #

1. 引入正确版本的Connector
   使用Elasticsearch 8.8.2对应Scala 2.12的Spark 3.x连接器包org.elasticsearch:elasticsearch-spark-30_2.12:8.8.2：

   • 用pyspark启动时指定依赖：

     pyspark --packages org.elasticsearch:elasticsearch-spark-30_2.12:8.8.2
     

   • 用spark-submit提交任务时：

     spark-submit --packages org.elasticsearch:elasticsearch-spark-30_2.12:8.8.2 your_script.py
     

2. 清理冲突依赖
   检查SPARK_HOME/jars目录，删除任何旧版本的Elasticsearch Spark Connector包，避免类加载冲突。

3. 优化写入配置（可选）
   针对Elasticsearch 8.x的特性，补充必要的配置参数：

   writer = (
       dataFrame.write.format("org.elasticsearch.spark.sql")
       .option("es.nodes", self.__host)
       .option("es.port", self.__port)
       .option("es.resource", f"{index}/{index}")
       .option("es.nodes.wan.only", "true")  # 远程集群需开启
       .option("es.index.auto.create", "false")  # 已手动管理索引，关闭自动创建
   )
   

内容的提问来源于stack exchange，提问作者Odai Tawalbeh