Airflow提交PySpark流任务至K8s Spark集群遇类型转换错误

环境信息 #

• Spark集群：Spark 3.3.0、PySpark 3.3、Scala 2.12.15、OpenJDK 64-Bit Server VM 11.0.15
• Airflow环境：Spark 3.1.2、PySpark 3.1.2、Scala 2.12.10、OpenJDK 64-Bit Server VM 1.8.0
• 依赖包：org.scala-lang:scala-library:2.12.8,org.apache.spark:spark-sql-kafka-0-10_2.12:3.3.0,org.apache.spark:spark-sql_2.12:3.3.0,org.apache.spark:spark-core_2.12:3.3.0,org.postgresql:postgresql:42.3.3

Airflow DAG代码 #

import airflow
from datetime import timedelta
from airflow import DAG
from time import sleep
from datetime import datetime
from airflow.providers.apache.spark.operators.spark_submit import SparkSubmitOperator

dag = DAG( dag_id = 'testpostgres.py', schedule_interval=None ,  start_date=datetime(2022, 1, 1), catchup=False)

spark_job = SparkSubmitOperator(application= '/usr/local/airflow/data/testpostgres.py',
                            conn_id= 'spark_kcluster',
                            task_id= 'spark_job_test',
                            dag= dag,
                            packages= "org.scala-lang:scala-library:2.12.8,org.apache.spark:spark-sql-kafka-0-10_2.12:3.3.0,org.apache.spark:spark-sql_2.12:3.3.0,org.apache.spark:spark-core_2.12:3.3.0,org.postgresql:postgresql:42.3.3",
                            conf ={
                                   'deploy-mode' : 'cluster',
                                   'executor_cores' : 1,
                                   'EXECUTORS_MEM' : '2G',
                                   'name' : 'spark-py',
                                   'spark.kubernetes.namespace' : 'sandbox',
                                   'spark.kubernetes.file.upload.path' : '/usr/local/airflow/data',
                                   'spark.kubernetes.container.image' : '**********',
                                   'spark.kubernetes.container.image.pullPolicy' : 'IfNotPresent',
                                   'spark.kubernetes.authenticate.driver.serviceAccountName' : 'spark',
                                   'spark.kubernetes.driver.volumes.persistentVolumeClaim.rwopvc.options.claimName' : 'data-pvc',
                                   'spark.kubernetes.driver.volumes.persistentVolumeClaim.rwopvc.mount.path' : '/usr/local/airflow/data',
                                   'spark.driver.extraJavaOptions' : '-Divy.cache.dir=/tmp -Divy.home=/tmp'
                                  }

)

PySpark任务代码 #

from pyspark.sql.functions import *
from pyspark.sql import SparkSession
from pyspark.sql import functions as F
from pyspark.sql.functions import dayofweek
from pyspark.sql.functions import date_format
from pyspark.sql.functions import hour
from functools import reduce
from pyspark.sql.types import DoubleType, StringType, ArrayType
import pandas as pd
import json

spark = SparkSession.builder.appName('spark').getOrCreate()


kafka_topic_name = '****'
kafka_bootstrap_servers = '*********' + ':' + '****'

streaming_dataframe = spark.readStream.format("kafka").option("kafka.bootstrap.servers", kafka_bootstrap_servers).option("subscribe", kafka_topic_name).option("startingOffsets", "earliest").load()
streaming_dataframe = streaming_dataframe.selectExpr("CAST(key AS STRING)", "CAST(value AS STRING)")

dataframe_schema = '******'
streaming_dataframe = streaming_dataframe.select(from_csv(col("value"), dataframe_schema).alias("pipeline")).select("pipeline.*")

tumblingWindows = streaming_dataframe.withWatermark("timeStamp", "48 hour").groupBy(window("timeStamp", "24 hour", "1 hour"), "phoneNumber").agg((F.first(F.col("duration")).alias("firstDuration")))

tumblingWindows = tumblingWindows.withColumn("start_window", F.col('window')['start'])
tumblingWindows = tumblingWindows.withColumn("end_window", F.col('window')['end'])
tumblingWindows = tumblingWindows.drop('window')

def postgres_write(tumblingWindows, epoch_id):
    tumblingWindows.write.jdbc(url=db_target_url, table=table_postgres, mode='append', properties=db_target_properties)

db_target_url = 'jdbc:postgresql://' + '*******'+ ':' + '****' + '/' + 'test'
table_postgres = '******'

db_target_properties = {
     'user': 'postgres',
     'password': 'postgres',
     'driver': 'org.postgresql.Driver'
}
query = tumblingWindows.writeStream.foreachBatch(postgres_write).start().awaitTermination()

错误日志 #

Driver stacktrace:
      at org.apache.spark.scheduler.DAGScheduler.failJobAndIndependentStages(DAGScheduler.scala:2672)
      at org.apache.spark.scheduler.DAGScheduler.$anonfun$abortStage$2(DAGScheduler.scala:2608)
      at org.apache.spark.scheduler.DAGScheduler.$anonfun$abortStage$2$adapted(DAGScheduler.scala:2607)
      at scala.collection.mutable.ResizableArray.foreach(ResizableArray.scala:62)
      at scala.collection.mutable.ResizableArray.foreach$(ResizableArray.scala:55)
      at scala.collection.mutable.ArrayBuffer.foreach(ArrayBuffer.scala:49)
      at org.apache.spark.scheduler.DAGScheduler.abortStage(DAGScheduler.scala:2607)
      at org.apache.spark.scheduler.DAGScheduler.$anonfun$handleTaskSetFailed$1(DAGScheduler.scala:1182)
      at org.apache.spark.scheduler.DAGScheduler.$anonfun$handleTaskSetFailed$1$adapted(DAGScheduler.scala:1182)
      at scala.Option.foreach(Option.scala:407)
      at org.apache.spark.scheduler.DAGScheduler.handleTaskSetFailed(DAGScheduler.scala:1182)
      at org.apache.spark.scheduler.DAGSchedulerEventProcessLoop.doOnReceive(DAGScheduler.scala:2860)
      at org.apache.spark.scheduler.DAGSchedulerEventProcessLoop.onReceive(DAGScheduler.scala:2802)
      at org.apache.spark.scheduler.DAGSchedulerEventProcessLoop.onReceive(DAGScheduler.scala:2791)
      at org.apache.spark.util.EventLoop$$anon$1.run(EventLoop.scala:49)
      at org.apache.spark.scheduler.DAGScheduler.runJob(DAGScheduler.scala:952)
      at org.apache.spark.SparkContext.runJob(SparkContext.scala:2228)
      at org.apache.spark.sql.execution.datasources.v2.V2TableWriteExec.writeWithV2(WriteToDataSourceV2Exec.scala:377)
      ... 42 more
Caused by: java.lang.ClassCastException: cannot assign instance of scala.collection.immutable.List$SerializationProxy to field org.apache.spark.sql.execution.datasources.v2.DataSourceRDDPartition.inputPartitions of type scala.collection.Seq in instance of org.apache.spark.sql.execution.datasources.v2.DataSourceRDDPartition
      at java.base/java.io.ObjectStreamClass$FieldReflector.setObjFieldValues(Unknown Source)
      at java.base/java.io.ObjectStreamClass$FieldReflector.checkObjectFieldValueTypes(Unknown Source)
      at java.base/java.io.ObjectStreamClass.checkObjFieldValueTypes(Unknown Source)
      at java.base/java.io.ObjectInputStream.defaultCheckFieldValues(Unknown Source)
      at java.base/java.io.ObjectInputStream.readSerialData(Unknown Source)
      at java.base/java.io.ObjectInputStream.readOrdinaryObject(Unknown Source)
      at java.base/java.io.ObjectInputStream.readObject0(Unknown Source)
      at java.base/java.io.ObjectInputStream.defaultReadFields(Unknown Source)
      at java.base/java.io.ObjectInputStream.readSerialData(Unknown Source)
      at java.base/java.io.ObjectInputStream.readOrdinaryObject(Unknown Source)
      at java.base/java.io.ObjectInputStream.readObject0(Unknown Source)
      at java.base/java.io.ObjectInputStream.readObject(Unknown Source)
      at java.base/java.io.ObjectInputStream.readObject(Unknown Source)
      at org.apache.spark.serializer.JavaDeserializationStream.readObject(JavaSerializer.scala:87)
      at org.apache.spark.serializer.JavaSerializerInstance.deserialize(JavaSerializer.scala:129)
      at org.apache.spark.executor.Executor$TaskRunner.run(Executor.scala:507)
      at java.base/java.util.concurrent.ThreadPoolExecutor.runWorker(Unknown Source)
      at java.base/java.util.concurrent.ThreadPoolExecutor$Worker.run(Unknown Source)
      at java.base/java.lang.Thread.run(Unknown Source)
Traceback (most recent call last):
File "/usr/local/airflow/data/spark-upload-d03175bc-8c50-4baf-8383-a203182f16c0/debug.py", line 20, in <module>
  streaming_dataframe.selectExpr("CAST(key AS STRING)", "CAST(value AS STRING)")\
File "/opt/spark/python/lib/pyspark.zip/pyspark/sql/streaming.py", line 107, in awaitTermination
File "/opt/spark/python/lib/py4j-0.10.9.5-src.zip/py4j/java_gateway.py", line 1321, in __call__
File "/opt/spark/python/lib/pyspark.zip/pyspark/sql/utils.py", line 196, in deco
pyspark.sql.utils.StreamingQueryException: Query [id = d0e140c1-830d-49c8-88b7-90b82d301408, runId = c0f38f58-6571-4fda-b3e0-98e4ffaf8c7a] terminated with exception: Writing job aborted
22/08/24 10:12:53 INFO SparkUI: Stopped Spark web UI at ************************
22/08/24 10:12:53 INFO KubernetesClusterSchedulerBackend: Shutting down all executors
22/08/24 10:12:53 INFO KubernetesClusterSchedulerBackend$KubernetesDriverEndpoint: Asking each executor to shut down
22/08/24 10:12:53 WARN ExecutorPodsWatchSnapshotSource: Kubernetes client has been closed.
22/08/24 10:12:53 INFO MapOutputTrackerMasterEndpoint: MapOutputTrackerMasterEndpoint stopped!
22/08/24 10:12:53 INFO MemoryStore: MemoryStore cleared
22/08/24 10:12:53 INFO BlockManager: BlockManager stopped
22/08/24 10:12:53 INFO BlockManagerMaster: BlockManagerMaster stopped
22/08/24 10:12:53 INFO OutputCommitCoordinator$OutputCommitCoordinatorEndpoint: OutputCommitCoordinator stopped!
22/08/24 10:12:54 INFO SparkContext: Successfully stopped SparkContext
22/08/24 10:12:54 INFO ShutdownHookManager: Shutdown hook called
22/08/24 10:12:54 INFO ShutdownHookManager: Deleting directory /var/data/spark-32ef85e0-e85c-4ac6-a46d-d3379ca58468/spark-adecf44a-dc60-4a85-bbe3-bc125f5cc39f/pyspark-f3ffaa5e-a490-464a-98d2-fbce223628eb
22/08/24 10:12:54 INFO ShutdownHookManager: Deleting directory /var/data/spark-32ef85e0-e85c-4ac6-a46d-d3379ca58468/spark-adecf44a-dc60-4a85-bbe3-bc125f5cc39f
22/08/24 10:12:54 INFO ShutdownHookManager: Deleting directory /tmp/spark-5acdd5e6-7f6e-45ec-adae-e98862e1537c

问题分析 #

核心错误为java.lang.ClassCastException，本质是跨版本序列化不兼容：

1. Airflow使用的Spark 3.1.2与集群的Spark 3.3.0版本差异，导致Spark内部类（如DataSourceRDDPartition）的序列化格式不一致；
2. Scala版本不匹配：Airflow环境Scala 2.12.10、依赖包指定2.12.8，与集群的2.12.15存在差异，Scala集合类的序列化代理无法兼容；
3. JDK版本差异：Airflow用JDK 8，集群用JDK 11，也可能加剧序列化问题。

解决方案 #

1. 统一Spark版本：将Airflow环境的Spark/PySpark升级至3.3.0，与集群版本完全一致，消除跨版本序列化差异；
2. 对齐Scala依赖版本：修改SparkSubmitOperator的packages参数，将org.scala-lang:scala-library:2.12.8改为2.12.15，确保所有Spark依赖的Scala版本与集群统一；
3. 统一JDK版本：将Airflow环境的JDK升级至11，与集群保持一致；
4. 优化写入容错：在foreachBatch的写入函数中添加异常捕获，避免单个批次失败导致整个流任务终止：

def postgres_write(tumblingWindows, epoch_id):
    try:
        tumblingWindows.write.jdbc(url=db_target_url, table=table_postgres, mode='append', properties=db_target_properties)
        print(f"批次 {epoch_id} 写入成功")
    except Exception as e:
        print(f"批次 {epoch_id} 写入失败: {str(e)}")
        # 可选：添加重试逻辑或记录失败数据

5. 检查镜像一致性：确保spark.kubernetes.container.image使用的镜像包含与集群一致的Spark、Scala和JDK版本，避免运行时环境差异。

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