Kafka消息Timestamp类型冲突致Delta Live Table流加载失败求助

Spark读取Kafka流时会自动加载元数据字段（如timestamp、topic等），而你的Kafka消息value中的JSON包含大小写不同的timeStamp字段。由于Spark SQL默认大小写不敏感，这两个字段被识别为同一字段，但类型分别是TimestampType和StringType，因此出现合并错误：

org.apache.spark.sql.AnalysisException: Failed to merge fields 'timeStamp' and 'timestamp'. Failed to merge incompatible data types StringType and TimestampType

解决方法 #

重命名Kafka自带的timestamp元数据字段，避免与消息内的timeStamp字段冲突，修改后的DLT代码如下：

from pyspark.sql.functions import col

@dlt.table(name = "tableOne", table_properties={"pipelines.reset.allowed":"false"})
def stream_bronze():
    kafka_df = (
        spark
        .readStream
        .format("kafka")
        .option("kafka.bootstrap.servers", "kafkaserver address")
        .option("subscribe","topic_name")
        .option("kafka.security.protocol", "SASL_SSL")
        .option("kafka.sasl.mechanism", "PLAIN")
        .option("kafka.sasl.jaas.config", """kafkashaded.org.apache.kafka.common.security.plain.PlainLoginModule required username="api key" password="secret";""")
        .load()
    )
    # 重命名Kafka元数据的timestamp字段，消除与消息内timeStamp的冲突
    return kafka_df.withColumnRenamed("timestamp", "kafka_timestamp")

如果需要解析消息中的JSON value，可以在重命名后继续处理，示例代码：

from pyspark.sql.functions import col, from_json
from pyspark.sql.types import StructType, StructField, StringType

# 根据你的实际消息结构定义schema
value_schema = StructType([
    StructField("timeStamp", StringType(), nullable=True),
    # 添加其他字段...
])

@dlt.table(name = "tableOne", table_properties={"pipelines.reset.allowed":"false"})
def stream_bronze():
    kafka_df = (
        spark
        .readStream
        .format("kafka")
        .option("kafka.bootstrap.servers", "kafkaserver address")
        .option("subscribe","topic_name")
        .option("kafka.security.protocol", "SASL_SSL")
        .option("kafka.sasl.mechanism", "PLAIN")
        .option("kafka.sasl.jaas.config", """kafkashaded.org.apache.kafka.common.security.plain.PlainLoginModule required username="api key" password="secret";""")
        .load()
        .withColumnRenamed("timestamp", "kafka_timestamp")
    )
    # 解析JSON格式的value字段
    return kafka_df.select(
        "kafka_timestamp", "topic", "partition", "offset", "key",
        from_json(col("value").cast("string"), value_schema).alias("value")
    )

原理说明 #

Spark SQL默认参数spark.sql.caseSensitive为false，对字段名大小写不敏感，因此timestamp和timeStamp会被判定为同一字段。通过重命名Kafka元数据的timestamp字段，让两个字段名称在Spark中被识别为不同字段，即可解决类型冲突问题。

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