We are migration one of our spark application from spark 3.0.3 to spark 3.2.2 with cassandra_connector 3.2.0 with Scala 2.12 version , and we are getting below exception
Caused by: org.apache.spark.SparkException: Job aborted due to stage failure: \
Task not serializable: java.io.NotSerializableException: \
org.apache.spark.sql.execution.datasources.v2.DataSourcePartitioning
Serialization stack:
- object not serializable (class: org.apache.spark.sql.execution.datasources.v2.DataSourcePartitioning, value: org.apache.spark.sql.execution.datasources.v2.DataSourcePartitioning@606805ac)
- field (class: org.apache.spark.sql.execution.adaptive.AQEShuffleReadExec, name: outputPartitioning, type: interface org.apache.spark.sql.catalyst.plans.physical.Partitioning)
- object (class org.apache.spark.sql.execution.adaptive.AQEShuffleReadExec, AQEShuffleRead local
+- ShuffleQueryStage 0
+- Exchange hashpartitioning(msisdn#29, record_creation_dttm#472, 200), ENSURE_REQUIREMENTS, [plan_id=85]
+- *(1) Filter (isnotnull(msisdn#29) AND isnotnull(record_creation_dttm#472))
Default spark optimized it to use broadcast join and application failed with serialization exception . But if I explicitly disable broadcast join application is working .
Note : On further digging down I found out that spark3.2.2 AQE and cassandra tuning parameter ["spark.cassandra.sql.inClauseToJoinConversionThreshold": "1"] creating this conflict, For now I resolved the issue with adding config "spark.sql.adaptive.autoBroadcastJoinThreshold": "-1"[Disabling AQE Broadcast join]
Sample code can be found here sample code