Spark Cassandra Connector 2.5.2自定义Map编解码器注册问题

问题根源 #

Spark-Cassandra-Connector 2.5.x 基于 Datastax Java Driver 4.x，和旧版2.0.x（依赖Driver 3.x）的API完全重构：

• 新版返回的CqlSession不再提供直接访问CodecRegistry的入口
• 手动创建独立的CqlSession或Cluster实例会与Spark连接器内部的Session隔离，且易因配置不匹配导致连接失败

推荐的注册方式 #

方式1：通过自定义SessionConfigFactory注入（官方推荐） #

通过Spark配置指定自定义的Session配置工厂，让连接器初始化时自动注册编解码器：

1. 配置SparkConf：

import org.apache.spark.SparkConf

val sparkConf = new SparkConf()
  .setAppName("CustomCassandraCodec")
  .set("spark.cassandra.connection.host", "127.0.0.1")
  .set("spark.cassandra.connection.local_dc", "datacenter1")
  // 指定自定义Session配置工厂
  .set("spark.cassandra.connection.config.factory", "com.your.package.CustomCodecSessionFactory")

2. 实现CustomCodecSessionFactory：

import com.datastax.spark.connector.cql.DefaultSessionConfigFactory
import com.datastax.oss.driver.api.core.CqlSessionBuilder

class CustomCodecSessionFactory extends DefaultSessionConfigFactory {
  override def configure(builder: CqlSessionBuilder): CqlSessionBuilder = {
    // 注册自定义Map编解码器
    super.configure(builder).addTypeCodecs(new CustomMapCodec())
  }
}

此方式保证编解码器被Spark连接器的所有Session实例共享，完全适配Spark的Cassandra操作。

方式2：全局CodecRegistry注册（临时方案） #

如果不想自定义工厂，可在Spark初始化前将编解码器注册到Driver的全局注册表：

import com.datastax.oss.driver.api.core.type.codec.TypeCodecs

// 在Spark连接Cassandra之前执行
TypeCodecs.DEFAULT_CODEC_REGISTRY.register(new CustomMapCodec())

注意：此方式会影响所有使用该Driver的代码，存在冲突风险，仅适合简单场景。

手动创建Session失败的排查 #

你遇到的AllNodesFailedException通常由以下原因导致：

• 本地数据中心名称错误：用nodetool status查看集群真实的DC名称
• 缺少认证配置：如果Cassandra开启了用户名密码认证，需添加.withAuthCredentials("user", "pass")
• 网络/防火墙问题：确保9042端口可访问，Cassandra服务正常运行

即使解决连接问题，手动创建的Session也无法用于Spark的Dataframe操作，因此不推荐这种方式。

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