Spark ML Pipelines（ML管道）

我会带着你远行 2021-09-19 02:14 359阅读 0赞

声明：代码主要以Scala为主，希望广大读者注意。本博客以代码为主，代码中会有详细的注释。相关文章将会发布在我的个人博客专栏[《Spark 2.0机器学习》][Spark 2.0]，欢迎大家关注。

--------------------

**一、Pipelines的主要概念**

ML可以应用于各种各样的数据类型，比如向量、文本、图形和结构化数据、API采用Spark SQL的DataFrame就是为了支持各种各样的数据类型。

1、Transformer（转换器）

转化器是特征变换和机器学习模型的抽象。转换器必须实现transform方法，这个方法将一个DataFrame转换成另一个DataFrame，通常是附加一个或者多个列。

2、Estimators（模型学习器）

模型学习器是拟合和训练数据的机器学习算法或者其他算法的抽象。模型学习器实现fit()方法，这个方法输入一个DataFrame并产生一个Model即一个Transformer（转换器）。

3、Pipeline（管道）

Pipeline将多个Transformers和Estimators绑在一起形成一个工作流。在机器学习中，通常会执行一系列算法来处理和学习模型，例如：一个简单的文本文档处理流程可能包括这几个步骤：首先把每个文档的文本分割成单词，然后将这些单词转换成一个数值型特征向量，最后使用特征向量和标签学习一个预测模型。MLlib代表一个流水线，就是一个Pipeline，Pipeline包含了一系列有特定顺序的管道步骤。

4、Parameter（参数）

MLlib的模型学习器和转换器使用同一的API来指定参数。Param是具有自包含定义的参数，ParamMap是一组（参数，值）对。

**二、Pipelines的实例**

一个pipeline由多个步骤组成，每一个步骤都是一个Transformer或者Estimator。这些步骤按顺序执行，首先输入DataFrame，然后通过每个阶段进行转换。在Transformer步骤中，DataFrame会调用transform()方法。在Estimator步骤中，fit()方法被调用并产生一个transformer，并且DataFrame会调用这个Transformer的transform()方法。

1、实例：Estimator, Transformer, Param

package sparkml
    
    import org.apache.log4j.{Level, Logger}
    import org.apache.spark.ml.classification.LogisticRegression
    import org.apache.spark.ml.linalg.{Vectors, Vector}
    import org.apache.spark.ml.param.ParamMap
    import org.apache.spark.sql.SparkSession
    import org.apache.spark.sql.Row
    
    /**
      * 实例：Estimator, Transformer, Param
      */
    
    object Pipeline {
      def main(args: Array[String]): Unit = {
        //设置日志输出格式
        Logger.getLogger("org").setLevel(Level.WARN)
    
        //定义SparkSession
        val spark = SparkSession.builder()
          .appName("pipeline")
          .master("local[*]")
          .getOrCreate()
    
        import spark.implicits._
    
        //1、训练样本
        val training = spark.createDataFrame(Seq(
          (1.0, Vectors.dense(0.0, 1.1, 0.1)),
          (0.0, Vectors.dense(2.0, 1.0, -1.0)),
          (0.0, Vectors.dense(2.0, 1.3, 1.0)),
          (1.0, Vectors.dense(0.0, 1.2, -0.5))
        )).toDF("label", "features")
    
        //2、创建逻辑回归Estimator
        val lr = new LogisticRegression()
    
        //3、通过setter方法设置模型参数
        lr.setMaxIter(10)  //设置最大迭代次数
          .setRegParam(0.01)  //设置正则迭代因子
    
        //4、训练模型
        val model1 = lr.fit(training)
    
        //5、通过ParamMap设置参数方法
        val paramMap = ParamMap(lr.maxIter -> 20)
          .put(lr.maxIter, 30)
          .put(lr.regParam -> 0.1, lr.threshold -> 0.55)
        //ParamMap合并
        /* val paramMap2 = paramMap(lr.probabilityCol -> "myProbability")
         val paramMapCombined = paramMap ++ paramMap2*/
    
        //6、训练模型，采用ParamMap参数
        //paramMapCombined会覆盖所有lr.set设置的参数
        //val model2 = lr.fit(training, paramMapCombined)
        val model2 = lr.fit(training, paramMap)
    
        //7、测试样本
        val test = spark.createDataFrame(Seq(
          (1.0, Vectors.dense(-1.0, 1.5, 1.3)),
          (0.0, Vectors.dense(3.0, 2.0, -0.1)),
          (1.0, Vectors.dense(0.0, 2.2, -1.5))
        )).toDF("label", "features")
    
        //8、对模型进行测试
        model2.transform(test)
          .select("features", "label", "prediction")
          .collect()
          .foreach{
            case Row(features: Vector, label: Double, prediction: Double) =>
              println(s"($features. $label) -> prediction=$prediction")
          }
      }
    
    }

结果如下所示：

([-1.0,1.5,1.3]. 1.0) -> prediction=1.0
    ([3.0,2.0,-0.1]. 0.0) -> prediction=0.0
    ([0.0,2.2,-1.5]. 1.0) -> prediction=1.0

2、实例：pipeline

package sparkml
    
    import org.apache.spark.ml.Pipeline
    import org.apache.spark.ml.classification.LogisticRegression
    import org.apache.spark.ml.feature.{HashingTF, Tokenizer}
    import org.apache.spark.sql.SparkSession
    import org.apache.spark.sql.Row
    import org.apache.spark.ml.linalg.Vector
    
    /**
      * 实例：Pipeline
      */
    
    object Pipeline2 {
      def main(args: Array[String]): Unit = {
        val spark = SparkSession.builder()
          .appName("pipeline2")
          .master("local[*]")
          .getOrCreate()
    
        import spark.implicits._
    
        //1.训练样本
        val training = spark.createDataFrame(Seq(
          (0L, "a b c d e spark", 1.0),
          (1L, "b d", 0.0),
          (2L, "spark f g h", 1.0),
          (3L, "hadoop mapreduce", 0.0)
        )).toDF("id", "text", "label")
    
        //2.ML pipeline参数设置，包括三个过程：首先是tokenizer，然后是hashingTF，最后是lr
        val tokenizer = new Tokenizer()
          .setInputCol("text")
          .setOutputCol("words")
        val hashingTF = new HashingTF()
          .setNumFeatures(1000)
          .setInputCol(tokenizer.getOutputCol)
          .setOutputCol("features")
        val lr = new LogisticRegression()
          .setMaxIter(10)
          .setRegParam(0.001)
        val pipeline = new Pipeline()
          .setStages(Array(tokenizer, hashingTF, lr))
    
        //3.训练pipeline模型
        val model = pipeline.fit(training)
    
        //4.保存pipeline模型
        // model.write.overwrite().save("E://temp//one")
    
        //5.保存pipeline
        // pipeline.write.overwrite().save("E://temp//two")
    
        //6.加载pipeline模型
        //val sameModel = PipelineModel.load("E://temp//one")
    
        //7.测试样本
        val test = spark.createDataFrame(Seq(
          (4L, "spark i j k"),
          (5L, "l m n"),
          (6L, "spark hadoop spark"),
          (7L, "apache hadoop")
        )).toDF("id", "text")
    
        //8.模型测试
        model.transform(test)
          .select("id", "text", "probability", "prediction")
          .collect()
          .foreach{
            case Row(id: Long, text: String, prob: Vector, prediction: Double) =>
              println(s"($id. $text) --> prob=$prob, prediction=$prediction")
          }
      }
    
    }

结果如下所示：

(4. spark i j k) --> prob=[0.15964077387874118,0.8403592261212589], prediction=1.0
    (5. l m n) --> prob=[0.8378325685476612,0.16216743145233875], prediction=0.0
    (6. spark hadoop spark) --> prob=[0.06926633132976273,0.9307336686702373], prediction=1.0
    (7. apache hadoop) --> prob=[0.9821575333444208,0.01784246665557917], prediction=0.0

3、实例：处理文本数据

文本内容如下所示：

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2dka3l4eTIwMTM_size_16_color_FFFFFF_t_70][]

测试代码如下：

package sparkml
    
    import org.apache.log4j.{Level, Logger}
    import org.apache.spark.ml.Pipeline
    import org.apache.spark.ml.classification.LogisticRegression
    import org.apache.spark.ml.feature.{HashingTF, Tokenizer}
    import org.apache.spark.sql.SparkSession
    import org.apache.spark.sql.functions._
    import org.apache.spark.sql.Row
    import org.apache.spark.ml.linalg.Vector
    
    /**
      * 实例：处理文本数据
      */
    
    object Pipeline3 {
      def main(args: Array[String]): Unit = {
        //设置日志输出格式
        Logger.getLogger("org").setLevel(Level.WARN)
    
        //1.定义SparkSession
        val spark = SparkSession.builder()
          .appName("textdata")
          .master("local[*]")
          .getOrCreate()
    
        import spark.implicits._
    
        //2.读取数据
        val data = spark.read.format("csv")
          .option("delimiter", "|")
          .option("header", "true")
          .load("C://Users//Machenike//Desktop//xzw//doc_class.dat")
          .toDF("myapp_id", "typenameid", "typename", "myapp_word", "myapp_word_all")
    
        val toDouble = udf[Double, String](_.toDouble)
        val clearData = udf[String, String](_.replace(" ", ""))
        val data2 = data.withColumn("label", toDouble(data("typenameid")))
          .withColumn("myapp_word_all", clearData(data("myapp_word_all")))
          .withColumnRenamed("myapp_word_all", "text")
          .withColumnRenamed("myapp_id", "id")
          .select("id", "text", "label")
    
        //3.将数据分为训练数据和测试数据
        val splitData = data2.randomSplit(Array(0.8, 0.2))
        val training = splitData(0)  //训练数据
        val testing = splitData(1)  //测试数据
    
        //4.tokenizer、hashingTF、lr
        val tokenizer = new Tokenizer()
          .setInputCol("text")
          .setOutputCol("words")
        val hashingTF = new HashingTF()
          .setNumFeatures(1000)
          .setInputCol(tokenizer.getOutputCol)
          .setOutputCol("features")
        val lr = new LogisticRegression()
          .setMaxIter(10)
          .setRegParam(0.001)
        //对tokenizer、hashingTF、lr进行封装
        val pipeline = new Pipeline()
          .setStages(Array(tokenizer, hashingTF, lr))
    
        //5.训练模型
        val model = pipeline.fit(training)
    
        //6.保存pipeline模型
        //model.write.overwrite().save("E://temp//one")
    
        //7.保存pipeline
        //pipeline.write.overwrite().save("E://temp//two")
    
        //8.加载pipeline模型
        //val sameModel = PipelineModel.load("E://temp//one")
    
        //9.测试
        val testing2 = testing.select("id", "text")
        model.transform(testing2)
          .select("id", "text", "probability", "prediction")
          .collect()
          .take(5)
          .foreach{
            case Row(id: String, text: String, prob: Vector, prediction: Double) =>
              println(s"($id, $text) --> prob = $prob, prediction = $prediction")
          }
      }
    
    }

结果如下图所示：（结果数据较多，此处只截取部分结果）

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2dka3l4eTIwMTM_size_16_color_FFFFFF_t_70 1][]

你们在此过程中遇到了什么问题，欢迎留言，让我看看你们都遇到了哪些问题。

[Spark 2.0]: https://blog.csdn.net/gdkyxy2013/column/info/37256
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2dka3l4eTIwMTM_size_16_color_FFFFFF_t_70]: /images/20210726/d8bc104ac07542be890d7e92214e8f51.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2dka3l4eTIwMTM_size_16_color_FFFFFF_t_70 1]: /images/20210726/4f04120a4ee34f76a439419b69686e31.png