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

Question

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

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 для ввода, и вложенные РДУ не поддерживаются в Спарк. Я выбрал этот подход, поскольку все эти наборы данных могут запускаться независимо друг от друга, и поэтому я хотел их распространять (следовательно, исключался цикл). Можете ли вы предложить, если я могу использовать другие типы (массивы, списки и т. Д.) Для ввода в качестве набора данных в LinearRegressionWithSGD или даже лучший подход, который по-прежнему будет распространять такие вычисления в Spark?

Answer 1

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

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