CDH6.3.2中hive-exec与commons-lang3冲突致Spark生成CSV失败

环境信息 #

• 本地开发：Talend Studio 7.3.1 补丁R2020-09
• 远程集群：Cloudera Hadoop 6.3.2（Spark 2.4.0）

报错堆栈 #

org.apache.spark.SparkException: Job aborted.
at org.apache.spark.sql.execution.datasources.FileFormatWriter$.write(FileFormatWriter.scala:198)
at org.apache.spark.sql.execution.datasources.InsertIntoHadoopFsRelationCommand.run(InsertIntoHadoopFsRelationCommand.scala:159)
at org.apache.spark.sql.execution.command.DataWritingCommandExec.sideEffectResult$lzycompute(commands.scala:104)
at org.apache.spark.sql.execution.command.DataWritingCommandExec.sideEffectResult(commands.scala:102)
at org.apache.spark.sql.execution.command.DataWritingCommandExec.doExecute(commands.scala:122)
at org.apache.spark.sql.execution.SparkPlan$$anonfun$execute$1.apply(SparkPlan.scala:131)
at org.apache.spark.sql.execution.SparkPlan$$anonfun$execute$1.apply(SparkPlan.scala:127)
at org.apache.spark.sql.execution.SparkPlan$$anonfun$executeQuery$1.apply(SparkPlan.scala:155)
at org.apache.spark.rdd.RDDOperationScope$.withScope(RDDOperationScope.scala:151)
at org.apache.spark.sql.execution.SparkPlan.executeQuery(SparkPlan.scala:152)
at org.apache.spark.sql.execution.SparkPlan.execute(SparkPlan.scala:127)
at org.apache.spark.sql.execution.QueryExecution.toRdd$lzycompute(QueryExecution.scala:80)
at org.apache.spark.sql.execution.QueryExecution.toRdd(QueryExecution.scala:80)
at org.apache.spark.sql.DataFrameWriter$$anonfun$runCommand$1.apply(DataFrameWriter.scala:668)
at org.apache.spark.sql.DataFrameWriter$$anonfun$runCommand$1.apply(DataFrameWriter.scala:668)
at org.apache.spark.sql.execution.SQLExecution$$anonfun$withNewExecutionId$1.apply(SQLExecution.scala:78)
at org.apache.spark.sql.execution.SQLExecution$.withSQLConfPropagated(SQLExecution.scala:125)
at org.apache.spark.sql.execution.SQLExecution$.withNewExecutionId(SQLExecution.scala:73)
at org.apache.spark.sql.DataFrameWriter.runCommand(DataFrameWriter.scala:668)
at org.apache.spark.sql.DataFrameWriter.saveToV1Source(DataFrameWriter.scala:276)
at org.apache.spark.sql.DataFrameWriter.save(DataFrameWriter.scala:270)
at org.apache.spark.sql.DataFrameWriter.save(DataFrameWriter.scala:228)
at org.apache.spark.sql.DataFrameWriter.csv(DataFrameWriter.scala:656)
at ... (private stuff)
Caused by: org.apache.spark.SparkException: Job aborted due to stage failure:
Aborting TaskSet 5.0 because task 6 (partition 6)
cannot run anywhere due to node and executor blacklist.
Most recent failure:
Lost task 4.2 in stage 5.0 (TID 225, slaaeizeba13.enelint.global, executor 4): java.io.InvalidClassException: org.apache.commons.lang3.time.FastDateParser; local class incompatible: stream classdesc serialVersionUID = 2, local class serialVersionUID = 3
at java.io.ObjectStreamClass.initNonProxy(ObjectStreamClass.java:616)
at java.io.ObjectInputStream.readNonProxyDesc(ObjectInputStream.java:1843)
at java.io.ObjectInputStream.readClassDesc(ObjectInputStream.java:1713)
at java.io.ObjectInputStream.readOrdinaryObject(ObjectInputStream.java:2000)
at java.io.ObjectInputStream.readObject0(ObjectInputStream.java:1535)
at java.io.ObjectInputStream.defaultReadFields(ObjectInputStream.java:2245)
at java.io.ObjectInputStream.readSerialData(ObjectInputStream.java:2169)
at java.io.ObjectInputStream.readOrdinaryObject(ObjectInputStream.java:2027)
at java.io.ObjectInputStream.readObject0(ObjectInputStream.java:1535)
at java.io.ObjectInputStream.defaultReadFields(ObjectInputStream.java:2245)
at java.io.ObjectInputStream.readSerialData(ObjectInputStream.java:2169)
at java.io.ObjectInputStream.readOrdinaryObject(ObjectInputStream.java:2027)
at java.io.ObjectInputStream.readObject0(ObjectInputStream.java:1535)
at java.io.ObjectInputStream.defaultReadFields(ObjectInputStream.java:2245)
at java.io.ObjectInputStream.readSerialData(ObjectInputStream.java:2169)
at java.io.ObjectInputStream.readOrdinaryObject(ObjectInputStream.java:2027)
at java.io.ObjectInputStream.readObject0(ObjectInputStream.java:1535)
at java.io.ObjectInputStream.defaultReadFields(ObjectInputStream.java:2245)
at java.io.ObjectInputStream.readSerialData(ObjectInputStream.java:2169)
at java.io.ObjectInputStream.readOrdinaryObject(ObjectInputStream.java:2027)
at java.io.ObjectInputStream.readObject0(ObjectInputStream.java:1535)
at java.io.ObjectInputStream.defaultReadFields(ObjectInputStream.java:2245)
at java.io.ObjectInputStream.readSerialData(ObjectInputStream.java:2169)
at java.io.ObjectInputStream.readOrdinaryObject(ObjectInputStream.java:2027)
at java.io.ObjectInputStream.readObject0(ObjectInputStream.java:1535)
at java.io.ObjectInputStream.defaultReadFields(ObjectInputStream.java:2245)
at java.io.ObjectInputStream.readSerialData(ObjectInputStream.java:2169)
at java.io.ObjectInputStream.readOrdinaryObject(ObjectInputStream.java:2027)
at java.io.ObjectInputStream.readObject0(ObjectInputStream.java:1535)
at java.io.ObjectInputStream.readObject(ObjectInputStream.java:422)
at org.apache.spark.serializer.JavaDeserializationStream.readObject(JavaSerializer.scala:75)
at org.apache.spark.serializer.JavaSerializerInstance.deserialize(JavaSerializer.scala:114)
at org.apache.spark.scheduler.ResultTask.runTask(ResultTask.scala:83)
at org.apache.spark.scheduler.Task.run(Task.scala:121)
at org.apache.spark.executor.Executor$TaskRunner$$anonfun$11.apply(Executor.scala:407)
at org.apache.spark.util.Utils$.tryWithSafeFinally(Utils.scala:1408)
at org.apache.spark.executor.Executor$TaskRunner.run(Executor.scala:413)
at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1142)
at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:617)
at java.lang.Thread.run(Thread.java:748)

问题场景 #

仅在调用tFileOutputDelimited组件的.csv方法时触发错误，生成的CSV用于后续迁移至Redshift数据库。

排查结论 #

hive-exec-2.1.1-cdh6.3.2内置的commons-lang3中FastDateParser的serialVersionUID为2，而作业依赖的commons-lang3-3.7中对应类的serialVersionUID为3，版本冲突导致Spark任务序列化/反序列化失败。

已尝试方案 #

• 修改CSV压缩格式
• 在Spark配置中设置自定义序列化器
• 在独立作业的pom中显式添加commons-lang3-3.7并排除hive-exec-2.1.1-cdh6.3.2中的commons-lang3依赖
• 尝试Stack Overflow相关问题的解决方案代码

提问 #

在不升级Talend补丁的情况下，还有哪些办法可以解决这个问题？

1. 强制统一commons-lang3版本 #

将作业依赖的commons-lang3版本改为与集群hive-exec内置版本一致（对应serialVersionUID=2的版本，如3.4）：

• 在Talend的模块视图中，找到commons-lang3依赖，修改版本为匹配集群的版本，并排除所有其他依赖传递过来的commons-lang3；
• 或在Talend作业的Spark运行参数中添加--jars，指定统一版本的commons-lang3.jar，让集群优先加载该jar覆盖内置版本。

2. 替换日期格式化逻辑，避免传递FastDateParser实例 #

自定义日期格式化逻辑，不依赖Talend组件默认的FastDateParser：

• 在写入CSV前，用Spark内置的date_format函数提前将日期字段转换为字符串：

  input_df.withColumn("formatted_date", functions.date_format(functions.col("date_col"), "yyyy-MM-dd HH:mm:ss"))
          .write()
          .option("header", "true")
          .csv("/target/path");
  

• 或改用Java原生SimpleDateFormat在本地完成日期转换，再传递字符串到集群任务。

3. 调整Spark类加载顺序 #

在Spark配置中添加以下参数，让用户提交的jar包类优先于集群内置类加载：

spark.driver.userClassPathFirst=true
spark.executor.userClassPathFirst=true

注意：此配置可能引发其他依赖冲突，需测试验证兼容性，可在Talend作业的Spark配置项中直接添加。

4. 隔离hive-exec的依赖传递 #

如果作业无需直接使用hive-exec功能，在Talend依赖管理中排除hive-exec的commons-lang3传递依赖；若必须使用hive-exec，则确保作业使用的commons-lang3版本兼容hive-exec的功能。

5. 绕过Talend组件，使用Spark原生CSV写入API #

直接在Talend的Java/Scala代码块中调用Spark DataFrame原生write.csv()方法，完全替代tFileOutputDelimited组件，避免Talend封装的FastDateParser实例被序列化到集群。

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