Spark DataFrame创建时TimestampType类型匹配错误求助

错误原因 #

报错明确指出：timestamp字段定义为TimestampType，但传入的是float类型数值（如161.657471），类型不匹配导致DataFrame创建失败。这说明你代码中custom_ml.results_df.rdd里的timestamp和created_time字段，并非你展示的时间字符串格式，而是被错误转换为了浮点数。

解决方案 #

1. 修正RDD中的时间字段类型 #

先明确浮点数对应的时间单位（秒/毫秒/微秒），将其转换为datetime对象后再创建DataFrame：

from pyspark.sql import Row
from datetime import datetime

def fix_time_fields(row):
    # 根据实际时间单位调整转换逻辑，示例假设是秒级时间戳
    # 如果是毫秒级，需改为：row.timestamp / 1000
    timestamp_dt = datetime.fromtimestamp(row.timestamp)
    created_time_dt = datetime.fromtimestamp(row.created_time)
    
    return Row(
        job_id=row.job_id,
        tag_name=row.tag_name,
        timestamp=timestamp_dt,
        avg_tag_value=row.avg_tag_value,
        result=row.result,
        created_time=created_time_dt
    )

# 转换RDD中的时间字段
corrected_rdd = custom_ml.results_df.rdd.map(fix_time_fields)

# 基于修正后的RDD创建DataFrame
_df_to_save = self.spark.createDataFrame(corrected_rdd, schema)

2. 直接从原DataFrame转换（无需转RDD） #

如果custom_ml.results_df本身还保留着原始时间字符串（如你展示的2023-02-15T21:45:00.000+0000），建议直接用原DataFrame转换，避免RDD的类型丢失：

from pyspark.sql.functions import col, to_timestamp
from pyspark.sql.types import DoubleType

_df_to_save = custom_ml.results_df \
    .withColumn("timestamp", to_timestamp(col("timestamp"), "yyyy-MM-dd'T'HH:mm:ss.SSSXXX")) \
    .withColumn("created_time", to_timestamp(col("created_time"), "yyyy-MM-dd'T'HH:mm:ss.SSSXXX")) \
    .withColumn("avg_tag_value", col("avg_tag_value").cast(DoubleType())) \
    .withColumn("result", col("result").cast(DoubleType()))

3. 先按浮点类型加载，再转换为时间类型 #

如果必须通过RDD创建，可以先临时用DoubleType加载时间字段，再转换为TimestampType：

from pyspark.sql.functions import to_timestamp
from pyspark.sql.types import StructType, StructField, StringType, DoubleType, TimestampType

# 临时Schema，将时间字段定义为DoubleType
temp_schema = StructType([
    StructField('job_id', StringType(), False),
    StructField('tag_name', StringType(), False),
    StructField('timestamp', DoubleType(), False),
    StructField('avg_tag_value', DoubleType(), True),
    StructField('result', DoubleType(), True),
    StructField('created_time', DoubleType(), False),
])

# 先创建临时DataFrame
temp_df = self.spark.createDataFrame(custom_ml.results_df.rdd, temp_schema)

# 转换时间字段类型
_df_to_save = temp_df \
    .withColumn("timestamp", to_timestamp(col("timestamp"))) \
    .withColumn("created_time", to_timestamp(col("created_time")))

关键排查步骤 #

如果不确定浮点数的含义，先打印RDD的前几条数据，确认字段的实际值和类型：

# 查看RDD的第一个元素，明确各字段的真实情况
print(custom_ml.results_df.rdd.take(1))

这能帮你确定浮点数是否为时间戳，以及对应的时间单位，避免转换出错误的时间。

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