转换 Leap Frame

无论是 MLeap 还是 Spark，Transformer 对于 Data Frame 的计算都是一种非常有用的抽象模型。让我们看看如何使用一个简单的 Transformer StringIndexer 来转换一帧 Data Frame。

  1. // Create a StringIndexer that knows how to index the two strings
  2. // In our leap frame
  3. val stringIndexer = StringIndexer(
  4.   shape = NodeShape().withStandardInput("a_string").withStandardOutput("a_string_index"),
  5.   model = StringIndexerModel(Seq("Hello, MLeap!", "Another row")))
  7. // Transform our leap frame using the StringIndexer transformer
  8. val indices = (for(lf <- stringIndexer.transform(leapFrame);
  9.                    lf2 <- lf.select("a_string_index")) yield {
  10.   lf2.dataset.map(_.getDouble(0))
  11. }).get.toSeq
  13. // Make sure our indexer did its job
  14. assert(indices == Seq(0.0, 1.0))

使用 Pipeline 来转换 Leap Frame

上面的例子可能不是很有趣。当你使用 Leap Frame 和 Transformer 一起来构建一个包含从原始特征到某些预测算法在内的完整 Pipeline 时，它们的真正威力才开始体现。让我们构造一个 Pipeline，其先通过 String Indexer 来生成索引，并把索引传给 One Hot Encoder，而后执行线性回归算法。 

  1. // Create our one hot encoder
  2. val oneHotEncoder = OneHotEncoder(shape = NodeShape.vector(1, 2, 
  3.                      inputCol = "a_string_index",
  4.                      outputCol = "a_string_oh"),
  5.   model = OneHotEncoderModel(2, dropLast = false))
  7. // Assemble some features together for use
  8. // By our linear regression
  9. val featureAssembler = VectorAssembler(
  10.      shape = NodeShape().withInput("input0", "a_string_oh").
  11.      withInput("input1", "a_double").withStandardOutput("features"),
  12.      model = VectorAssemblerModel(Seq(TensorShape(2), ScalarShape())))
  14. // Create our linear regression
  15. // It has two coefficients, as the one hot encoder
  16. // Outputs vectors of size 2
  17. val linearRegression = LinearRegression(shape = NodeShape.regression(3),
  18.   model = LinearRegressionModel(Vectors.dense(2.0, 3.0, 6.0), 23.5))
  20. // Create a pipeline from all of our transformers
  21. val pipeline = Pipeline(
  22.       shape = NodeShape(),
  23.       model = PipelineModel(Seq(stringIndexer, oneHotEncoder, featureAssembler, linearRegression)))
  25. // Transform our leap frame using the pipeline
  26. val predictions = (for(lf <- pipeline.transform(leapFrame);
  27.                        lf2 <- lf.select("prediction")) yield {
  28.   lf2.dataset.map(_.getDouble(0))
  29. }).get.toSeq
  31. // Print our predictions
  32. //   > 365.70000000000005
  33. //   > 166.89999999999998
  34. println(predictions.mkString("\n"))

这个任务体现了 MLeap 的意义在于执行我们通过 Spark、PySpark、Scikit-Learn 或者 Tensorflow 等机器学习框架训练得到的 Pipeline。