将转换应用于新数据 Spark

Question

我正在使用 Spark (core/mlib) 和 Java，版本 2.3.1。

我正在对数据集应用三个转换 - StringIndexer、OneHotEncoderEstimator、VectorAssember - 这是为了将我的数据集中的分类变量转换为每个类别的 1 和 0 的单独列。在我的火车数据上，这种转换没有问题，一切都符合预期，我正在将此模型保存到文件中。

当我尝试在新数据点上使用此模型时出现问题：

public static double loadModel(Obj newData) {
    SparkSession spark = Shots.buildSession();

    //Function which applies transformations
    Dataset<Row> data = buildDataset(spark, Arrays.asList(newData));

    LogisticRegressionModel lrModel = LogisticRegressionModel.load(modelPath);
    //Error is thrown here as the model doesn't seem to understand the input
    Dataset<Row> preds = lrModel.transform(data);
    preds.show();
}

我认为，问题在于转换现在仅应用于一行数据，该数据仅输出一个类别的分类特征和一个转换后只有一个元素的向量。当应用 LogisticRegressionModel 变换时，这会导致错误，这需要一个长度大于该特征的向量......我认为。

我知道我的错误是不知道如何将火车变换应用于新数据...但我不确定错误到底在哪里，因此不知道在哪里可以找到答案（是问题​​所在保存模型后，我是否需要保存其他东西，例如管道等）。

实际抛出的错误是 -

java.lang.IllegalArgumentException: requirement failed: BLAS.dot(x: Vector, y:Vector) was given Vectors with non-matching sizes: x.size = 7, y.size = 2 - the reason why I have come to the conclusions above is a visual examination of the data.

一个例子可能有助于解释：我有一个具有 3 个值 [Yes, No, Maybe] 的分类特征。我的火车数据包含所有三个值，我最终得到一个长度为 3 的向量特征来表示类别。

然后我在单个数据点上使用相同的管道来预测一个值，但分类特征只能是 Yes、No 或 Maybe，因为只有一个数据点。因此，当您应用与上述相同的转换时，您最终会得到一个包含一个元素而不是三个元素的向量，从而导致模型转换引发错误。

Answer 1

你有两个选择：

1. handle_data_PipelineModel ==> df ---> split_dataset ==> train_df/test_df--> arithmetic_PipelineModel----->test_model--->evaluate

2. df == > split_dataset ==> train_df/test_df--> PipelineModel(handle_data_stage and arithmetic_stage) ---> probably error

选项 1 是安全的：您需要保存 handle_data_PipelineModel 和 arithmetic_PipelineModel。

选项 2 不好：无论您如何保存模型。当你先拆分数据时，train_df和test_df的分布会发生变化。

注意：分割后的数据集一定不能在PipelineModel之前处理数据。

Answer 2

我也面临同样的问题。 StringIndexer 对于测试中的新值或新数据将失败，因此我们可以选择跳过那些未知值。

new StringIndexer().setHandleInvalid("skip")

或将训练数据和测试数据的联合传递到管道并在转换后对其进行拆分。

Answer 3

这里的问题是我将模型保存在错误的位置。

要保留先前转换的效果，您需要将管道拟合到数据然后写入/保存模型。这意味着保存 PipelineModel 而不是 Pipeline。如果您在加载数据后适合，则转换将完全重新应用，您将失去转换工作所需的状态。

Answer 4

一般来说，您没有正确使用 API。正确的工作流程应包括保留在此过程中训练的一整套Models（在您的情况下至少为StringIndexerModel，其他组件看起来像Transformers）并将这些重新应用于新数据。

最方便的方法是使用Pipeline：

 val pipeline = new Pipeline().setStages(Arrray(indexer, encoder, assembler, lr))
 val pipelineModel = pipeline.fit(data)

 pipelineModel.transform(data)

PipelineModels 可以保存为任何其他组件，只要它的所有阶段都是可写的。