apache-spark - Spark ML - 从新数据元素创建特征向量以进行预测

Question

tl;博士

我在 Spark 2.10 中拟合了 LinearRegression 模型 - 使用 StringIndexer 和 OneHotEncoder 后，我有一个 ~44 元素特征向量。对于我想要预测的新数据，如何从新数据元素创建特征向量？

更多详细信息

首先，这是一个完全人为的示例，用于学习如何执行此操作。使用带有字段的日志:

"elapsed_time", "api_name", "method", and "status_code"

我们将创建标签elapsed_time的模型，并使用其他字段作为我们的特征集。下面将分享完整的代码。

步骤 - 精简

1. 将数据读入 DataFrame
2. 使用 StringIndexer 为我们的每个功能建立索引
3. 使用 OneHotEncoder OneHotEncode 索引功能
4. 使用 VectorAssembler 创建我们的特征向量
5. 将数据分为训练集和测试集
6. 拟合模型并根据测试数据进行预测

结果很糟糕，但就像我说的，这是一个人为的练习......

我需要学习如何做

例如，如果一个新的日志条目进入流应用程序，我将如何从新数据创建特征向量并将其传递给 Predict()？

新的日志条目可能是:

{api_name":"/sample_api_1/v2","method":"GET","status_code":"200","elapsed_time":39}

后向量汇编器

状态代码向量

(14,[0],[1.0])

API_名称_向量

(27,[0],[1.0])

方法向量

(3,[0],[1.0])

特征向量

(44,[0,14,41],[1.0,1.0,1.0])

乐码

%spark

import org.apache.spark.ml.feature.{OneHotEncoder, StringIndexer, VectorAssembler, StringIndexerModel, VectorSlicer}
import org.apache.spark.ml.linalg.Vectors
import org.apache.spark.ml.regression.LinearRegression
import org.apache.spark.sql.DataFrame

val logs = sc.textFile("/Users/z001vmk/data/sample_102M.txt")
val dfLogsRaw: DataFrame = spark.read.json(logs)

val dfLogsFiltered = dfLogsRaw.filter("status_code != 314").drop("extra_column")

// Create DF with our fields of concern.
val dfFeatures: DataFrame = dfLogsFiltered.select("elapsed_time", "api_name", "method", "status_code")

// Contrived goal:
// Use elapsed time as our label given features api_name, status_code, & method.
// Train model on small (100Mb) dataset
// Be able to predict elapsed_time given a new record similar to this example:
// --> {api_name":"/sample_api_1/v2","method":"GET","status_code":"200","elapsed_time":39}

// Indexers
val statusCodeIdxr: StringIndexer = new StringIndexer().setInputCol("status_code").setOutputCol("status_code_idx").setHandleInvalid("skip")
val apiNameIdxr: StringIndexer = new StringIndexer().setInputCol("api_name").setOutputCol("api_name_idx").setHandleInvalid("skip")
val methodIdxr: StringIndexer = new StringIndexer().setInputCol("method").setOutputCol("method_idx").setHandleInvalid("skip")
// Index features:
val dfIndexed0: DataFrame = statusCodeIdxr.fit(dfFeatures).transform(dfFeatures)
val dfIndexed1: DataFrame = apiNameIdxr.fit(dfIndexed0).transform(dfIndexed0)
val indexed: DataFrame = methodIdxr.fit(dfIndexed1).transform(dfIndexed1)
// OneHotEncoders
val statusCodeEncoder: OneHotEncoder = new OneHotEncoder().setInputCol(statusCodeIdxr.getOutputCol).setOutputCol("status_code_vec")
val apiNameEncoder: OneHotEncoder = new OneHotEncoder().setInputCol(apiNameIdxr.getOutputCol).setOutputCol("api_name_vec")
val methodEncoder: OneHotEncoder = new OneHotEncoder().setInputCol(methodIdxr.getOutputCol).setOutputCol("method_vec")
// Encode feature vectors
val encoded0: DataFrame = statusCodeEncoder.transform(indexed)
val encoded1: DataFrame = apiNameEncoder.transform(encoded0)
val encoded: DataFrame = methodEncoder.transform(encoded1)
// Limit our dataset to necessary elements:
val dataset0 = encoded.select("elapsed_time", "status_code_vec", "api_name_vec", "method_vec").withColumnRenamed("elapsed_time", "label")

// Assemble feature vectors
val assembler: VectorAssembler = new VectorAssembler().setInputCols(Array("status_code_vec", "api_name_vec", "method_vec")).setOutputCol("features")
val dataset1 = assembler.transform(dataset0)
dataset1.show(5,false)
// Prepare the dataset for training (optional):
val dataset: DataFrame = dataset1.select("label", "features")
dataset.show(3,false)

val Array(training, test) = dataset.randomSplit(Array(0.8, 0.2))

// Create our Linear Regression Model
val lr: LinearRegression = new LinearRegression().setMaxIter(10).setRegParam(0.3).setElasticNetParam(0.8).setLabelCol("label").setFeaturesCol("features")
val lrModel = lr.fit(training)

val predictions = lrModel.transform(test)
predictions.show(20,false)

如果您有兴趣，可以将这些全部粘贴到 Zeppelin 笔记本中。

总结

所以，我一直在寻找的是如何将新数据转换为大约 35 个元素的特征向量，并使用适合训练数据的模型来转换它并获得预测。我怀疑模型本身保存有元数据，或者在这种情况下需要从 StringIndexers 维护元数据 - 但这就是我找不到的。

很高兴能够找到文档或示例 - 感谢所有帮助。

谢谢!

Answer 1

简短回答:管道模型。

不过，为了确保您理解，如果不需要，您不想在启动应用程序时创建模型。除非您要使用数据集和反馈，否则这是愚蠢的。在 Spark Submit session 中创建模型(或使用 Zeppelin 等笔记本 session )并将其保存。这就是你的数据科学。

大多数 DS 人员都会将模型交给 DevOps/数据工程师使用。他们所要做的就是在对象加载到内存后调用 .predict() 。