Spark Structured Streaming处理parquet时间戳实现1分钟滚动聚合的报错问题

1. 直接触发MatchError的原因 #

代码中通过new TimestampType()创建类型实例的写法完全错误。
Spark SQL所有内置数据类型都是Scala端实现的单例对象，Java代码开发时必须引用org.apache.spark.sql.types.DataTypes类下预定义的静态常量实例，不允许自行通过new创建。自行创建的TimestampType实例不被Spark内部的RowEncoder类型匹配逻辑识别，在编码器初始化阶段匹配失败时就会抛出遇到的MatchError异常。

2. 时间戳读取与窗口函数的兼容问题 #

之前将timestamp字段定义为Long类型读取时窗口函数报错，是因为window滚动窗口函数的输入硬要求是Spark SQL的TimestampType字段（底层对应java.sql.Timestamp类型），不支持Long格式的Unix时间戳直接作为输入。
结合数据特征——parquet中存储的是毫秒级Long时间戳，直接在读取schema时强制指定为TimestampType也会出现解码不兼容的问题，因为parquet存储Long值的二进制结构和TimestampType的解码规则不匹配。

1. 按照parquet实际存储的物理结构定义读取schema，所有DataType全部引用预定义常量，禁止自行实例化
2. 流式读取完成后，将毫秒级Long时间戳转换为Spark支持的TimestampType
3. 基于转换后的时间字段完成窗口聚合

注意：Spark内置Long转Timestamp的规则是将Long值识别为秒级Unix时间戳，如果字段存的是毫秒级数值，需要先除以1000再做类型转换，否则会出现时间量级错误。

import org.apache.spark.sql.types.DataTypes;
import static org.apache.spark.sql.functions.*;

// 正确构建schema：全部使用预定义DataType常量
StructType readSchema = new StructType()
        .add("value", DataTypes.IntegerType)
        .add("deviceId", DataTypes.LongType)
        // 匹配parquet实际存储的毫秒Long类型
        .add("timestamp", DataTypes.LongType);

// 流式读取parquet
Dataset<Row> inputDataFrame = sparkSession.readStream()
        .schema(readSchema)
        .parquet(path);

// 时间戳类型转换：毫秒Long -> 秒 -> TimestampType
Dataset<Row> converted = inputDataFrame.withColumn(
        "event_time",
        col("timestamp").divide(1000).cast(DataTypes.TimestampType)
);

// 1分钟滚动窗口聚合
Dataset<Row> aggregations = converted.groupBy(
        window(col("event_time"), "1 minute"),
        col("deviceId")
).agg(sum("value").alias("device_total_value"));

• 窗口时间粒度的标准写法为单数单位形式，如"1 minute"，原写法"1 minutes"虽然被Spark兼容，但不符合官方语法规范
• 生产环境建议基于转换后的event_time字段配置水位线，例如converted.withWatermark("event_time", "10 minutes")，避免聚合状态无限制膨胀，适配多设备并行上报带来的数据乱序场景

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