可以在Spark批处理上创建模型并在Spark流中使用它吗？

Question

我可以在Spark批处理中创建模型并将其用于Spark流式传输以进行实时处理吗？

我已经看到Apache Spark网站上的各种示例,其中训练和预测都建立在相同类型的处理(线性回归)上.

Answer 1

我可以在Spark批处理中创建模型并将其用于Spark流式传输以进行实时处理吗？

当然,是的.在spark社区,他们称之为离线培训在线预测.spark中的许多训练算法允许您将模型保存在文件系统HDFS/S3上.流应用程序可以加载相同的模型.您只需调用模型的预测方法来进行预测.

请参阅此链接中的 Streaming + MLLib部分.

例如,如果您想离线训练DecisionTree并在线进行预测...

批量申请 -

    val model = DecisionTree.trainClassifier(trainingData, numClasses, categoricalFeaturesInfo,impurity, maxDepth, maxBins)
    model.save(sc, "target/tmp/myDecisionTreeClassificationModel")

在流媒体应用中 -

    val sameModel = DecisionTreeModel.load(sc, "target/tmp/myDecisionTreeClassificationModel")
    sameModel.predict(newData)

Answer 2

这是我刚刚实施的另一种解决方案。

我在 Spark-Batch 中创建了一个模型。假设最终的模型对象名称是 regmodel。

final LinearRegressionModel regmodel =algorithm.run(JavaRDD.toRDD(parsedData));

Spark 上下文名称是 sc as

JavaSparkContext sc = new JavaSparkContext(sparkConf);

现在，在相同的代码中，我使用相同的 sc 创建 Spark 流

final JavaStreamingContext jssc = new JavaStreamingContext(sc,new Duration(Integer.parseInt(conf.getWindow().trim())));

并进行这样的预测：

JavaPairDStream<Double, Double> predictvalue = dist1.mapToPair(new PairFunction<LabeledPoint, Double,Double>() {
                private static final long serialVersionUID = 1L;
                @Override
                public Tuple2<Double, Double> call(LabeledPoint v1) throws Exception {
                    Double p = v1.label();
                    Double q = regmodel.predict(v1.features());
                    return new Tuple2<Double, Double>(p,q);
                }
            });