我的用例:

从MongoDB集合中读取以下形式的数据:

{    "_id" : ObjectId("582cab1b21650fc72055246d"),    "label" : 167.517838916715,    "features" : [         10.0964787450654,         218.621137772497,         18.8833848806122,         11.8010251302327,         1.67037687829152,         22.0766170950477,         11.7122322171201,         12.8014773524475,         8.30441804118235,         29.4821268054137    ]}

并将其传递给org.apache.spark.ml.regression.LinearRegression类以创建预测模型。

我的问题:

Spark连接器读取的”features”是Array[Double]类型。
LinearRegression.fit(…)期望一个包含标签列和特征列的DataSet。
特征列必须是VectorUDT类型（因此DenseVector或SparseVector都可以使用）。
我无法将Array[Double]类型的features映射到DenseVector，因为没有相关的编码器:

Error:(23, 11) 无法为存储在数据集中的类型找到编码器。 通过导入spark.implicits._支持基本类型（Int, String等）和产品类型（案例类）。 将来版本将增加对序列化其他类型的支持。  .map{case Row(label: Double, features: Array[Double]) => Row(label, Vectors.dense(features))}

无法定义自定义编码器。

我的问题:

• 有没有办法设置Spark连接器的配置，以Dense/SparseVector的形式读取”features”数组?
• 有没有其他方法可以实现这一点（例如，不使用中间.csv文件并使用libsvm加载）?

我的代码:

import com.mongodb.spark.MongoSparkimport org.apache.spark.ml.linalg.Vectorsimport org.apache.spark.ml.regression.LinearRegressionimport org.apache.spark.sql.{Row, SparkSession}case class DataPoint(label: Double, features: Array[Double])object LinearRegressionWithMongo {  def main(args: Array[String]) {    val spark = SparkSession      .builder()      .appName("LinearRegressionWithMongo")      .master("local[4]")      .config("spark.mongodb.input.uri", "mongodb://127.0.0.1/LinearRegressionTest.DataPoints")      .getOrCreate()    import spark.implicits._    val dataPoints = MongoSpark.load(spark)      .map{case Row(label: Double, features: Array[Double]) => Row(label, Vectors.dense(features))}    val splitData = dataPoints.randomSplit(Array(0.7, 0.3), 42)    val training = splitData(0)    val test = splitData(1)    val linearRegression = new LinearRegression()      .setLabelCol("label")      .setFeaturesCol("features")      .setRegParam(0.0)      .setElasticNetParam(0.0)      .setMaxIter(100)      .setTol(1e-6)    // 训练模型    val startTime = System.nanoTime()    val linearRegressionModel = linearRegression.fit(training)    val elapsedTime = (System.nanoTime() - startTime) / 1e9    println(s"训练时间: $elapsedTime 秒")    // 打印线性回归的权重和截距。    println(s"权重: ${linearRegressionModel.coefficients} 截距: ${linearRegressionModel.intercept}")    val modelEvaluator = new ModelEvaluator()    println("训练数据结果:")    modelEvaluator.evaluateRegressionModel(linearRegressionModel, training, "label")    println("测试数据结果:")    modelEvaluator.evaluateRegressionModel(linearRegressionModel, test, "label")    spark.stop()  }}

任何帮助都将非常感激！

回答：

对此有一个快速修复。如果数据已加载到名为df的DataFrame中，其中包含:

• id – SQL double类型。
• features – SQL array<double>类型。

像这样

val df = Seq((1.0, Array(2.3, 3.4, 4.5))).toDF("id", "features")

你可以select你需要用于后续处理的列:

val idAndFeatures = df.select("id", "features")

转换为静态类型的Dataset:

val tuples = idAndFeatures.as[(Double, Seq[Double])]

map并转换回Dataset[Row]:

val spark: SparkSession = ???import spark.implicits._import org.apache.spark.ml.linalg.Vectorstuples.map { case (id, features) =>   (id, Vectors.dense(features.toArray))}.toDF("id", "features")

你可以在这里找到与你当前方法相比的详细解释。