【问题标题】:Spark ml streaming predictOnValues how to save results?Spark ml流predictOnValues如何保存结果?
【发布时间】:2018-02-17 10:57:25
【问题描述】:
我有以下代码:
StreamingLinearRegressionWithSGD regressionWithSGD =
new StreamingLinearRegressionWithSGD()
.setInitialWeights(Vectors.zeros(featuresNumber));
JavaDStream<LabeledPoint> trainingData = streamingContext.textFileStream(model.getTrainPath()).map(LabeledPoint::parse).cache();
JavaDStream<LabeledPoint> testData = streamingContext.textFileStream(model.getPredictPath()).map(LabeledPoint::parse);
regressionWithSGD.trainOn(trainingData);
regressionWithSGD.predictOnValues(testData.mapToPair(lp -> new Tuple2<>(lp.label(), lp.features()))).print();
我想把结果放到一些文件/数据库/队列等而不是print() 可以吗?
【问题讨论】:
标签:
java
apache-spark
spark-streaming
apache-spark-ml
【解决方案1】:
我想通了
StreamingLinearRegressionWithSGD regressionWithSGD =
new StreamingLinearRegressionWithSGD()
.setInitialWeights(Vectors.zeros(featuresNumber));
JavaDStream<LabeledPoint> trainingData = streamingContext.textFileStream(model.getTrainPath()).map(LabeledPoint::parse).cache();
JavaDStream<LabeledPoint> testData = streamingContext.textFileStream(model.getPredictPath()).map(LabeledPoint::parse);
regressionWithSGD.trainOn(trainingData);
JavaDStream<Double> doubleJavaDStream=regressionWithSGD.predictOn(testData.map(labeledPoint -> labeledPoint.features()));
doubleJavaDStream.dstream().saveAsTextFiles("result","out");
因此,我们得到了 result-{timestamp}.out 文件夹。