Spark Scala中TimestampType不被VectorAssembler支持的问题求助

你遇到的这个报错很典型——VectorAssembler只支持数值型、布尔型和向量型的列，而TimestampType属于日期时间类型，不在它的支持范围内，所以直接把time列放进特征向量里会触发这个IllegalArgumentException。

解决方案：把Timestamp转换成可用于建模的数值特征 #

我们需要把时间戳转换成数值形式，常见的两种实用方式：

1. 将Timestamp转换为从固定时间点开始的秒数/毫秒数（比如Unix时间戳），得到纯数值列，直接适配线性回归的数值特征要求
2. 提取时间维度的特征（比如小时、分钟、星期几、月份等），这些离散特征可进一步处理为数值型（比如独热编码）

下面针对第一种最直接的方案修改你的代码：

object regressionLinear { 
  case class X( time:String,nodeID: Int, posX: Double,posY: Double, speed: Double,period: Int) 
  def main (args: Array[String]) { 
    Logger.getLogger("org").setLevel(Level.OFF) 
    Logger.getLogger("akka").setLevel(Level.OFF) 

    /** * Read the input data */ 
    var dataset = "C:\\spark\\A6-d07-h08.csv" 
    if (args.length > 0) { 
      dataset = args(0) 
    } 

    val spark = SparkSession 
      .builder 
      .appName("regressionsol") 
      .master("local[4]") 
      .getOrCreate() 

    import spark.implicits._
    import org.apache.spark.sql.functions._

    val data = spark.sparkContext.textFile(dataset) 
      .map(line=>line.split(",")) 
      .map(userRecord => (
        userRecord(0).trim.toString, 
        userRecord(1).trim.toInt, 
        userRecord(2).trim.toDouble,
        userRecord(3).trim.toDouble, 
        userRecord(4).trim.toDouble,
        userRecord(5).trim.toInt
      )) 
      .toDF("time","nodeID","posX", "posY","speed","period")
      // 先转换为Timestamp类型，再转成Unix时间戳（秒数，纯数值型）
      .withColumn("time_ts", $"time".cast("timestamp"))
      .withColumn("time_unix", unix_timestamp($"time_ts"))
      // 删除原时间列和中间Timestamp列，避免混淆
      .drop("time", "time_ts")

    val assembler = new VectorAssembler() 
      // 用转换后的数值型time_unix代替原time列
      .setInputCols( Array( "time_unix","nodeID","posX", "posY","speed","period")) 
      .setOutputCol("features") 

    val lr = new LinearRegression() 
      .setLabelCol("period") 
      .setFeaturesCol("features") 
      .setRegParam(0.1) 
      .setMaxIter(100) 
      .setSolver("l-bfgs") 

    val steps = Array(assembler, lr) 
    val pipeline = new Pipeline() 
      .setStages(steps) 

    val Array(training, test) = data.randomSplit(Array(0.75, 0.25), seed = 12345) 
    val model = pipeline.fit(training) 
    val holdout = model.transform(test) 
    holdout.show(20) 

    val prediction = holdout.select("prediction", "period","nodeID")
      .orderBy(abs(col("prediction")-col("period"))) 
    prediction.show(20) 

    val rm = new RegressionMetrics(prediction.rdd.map{ x => (x(0).asInstanceOf[Double], x(1).asInstanceOf[Double]) }) 
    println(s"RMSE = ${rm.rootMeanSquaredError}") 
    println(s"R-squared = ${rm.r2}") 

    spark.stop() 
  } 
}

补充说明 #

• 如果你的时间格式不是Spark默认支持的格式，需要在unix_timestamp中指定格式，比如unix_timestamp($"time", "yyyy-MM-dd HH:mm:ss")
• 若想提取时间维度特征（比如小时、星期几），可以用hour($"time_ts")、dayofweek($"time_ts")等函数生成新的数值列，再加入到VectorAssembler的输入列中
• 线性回归对数值特征的尺度敏感，建议后续可以加入StandardScaler对特征做标准化处理，进一步提升模型效果

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