我的用例:
从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– SQLdouble类型。features– SQLarray<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")你可以在这里找到与你当前方法相比的详细解释。
