Линейная регрессия на Apache Spark

Question

У нас есть ситуация, когда мы должны запустить линейную регрессию для миллионов небольших наборов данных и сохранить веса и перехват для каждого из этих наборов данных. Для этого я написал приведенный ниже скалярный код, в котором я подал каждый из этих наборов данных в виде строки в СДР, а затем пытаюсь запустить регрессию для каждого (данные - это СДР, в котором (метка, элементы) хранятся в каждом из них. строка, в этом случае у нас есть одна особенность на ярлык):

val x = data.flatMap { line => line.split(' ')}.map { line =>
  val parts = line.split(',')
  val parsedData1 = LabeledPoint(parts(0).toDouble, Vectors.dense(parts(1).split(' ').map(_.toDouble)))
  val model = LinearRegressionWithSGD.train(sc.parallelize(List(parsedData1)),100)//using parallelize to convert data to type RDD
  (model.intercept,model.weights)
}

Проблема здесь в том, что LinearRegressionWithSGD ожидает RDD для ввода, и вложенные RDD не поддерживаются в Spark. Я выбрал этот подход, так как все эти наборы данных могут работать независимо друг от друга, и поэтому я хотел распространять их (следовательно, исключил зацикливание). Не могли бы вы порекомендовать, могу ли я использовать другие типы (массивы, списки и т. Д.) Для ввода в качестве набора данных в LinearRegressionWithSGD или даже в качестве лучшего подхода, который все равно будет распространять такие вычисления в Spark?

Answer 1

val modelList = for {item <- dataSet} yield {
    val data = MLUtils.loadLibSVMFile(context, item).cache()
    val model = LinearRegressionWithSGD.train(data)
    model
}

Может быть, вы можете разделить свои входные данные на несколько файлов и хранить в HDFS. Используйте каталог этих файлов в качестве входных данных, вы можете получить список моделей.