Spark Pipeline官方文档

ML Pipelines(译文)

官方文档链接:https://spark.apache.org/docs/latest/ml-pipeline.html

概述

在这一部分,我们将要介绍ML Pipelines,它提供了基于DataFrame上统一的高等级API,可以帮助使用者创建和调试机器学习工作流;

目录:

  • Pipelines中主要的概念:
    • DataFrame
    • Pipeline组件
      • Transformers:转换器
      • Estimators:预测器
      • Pipelines组件属性
    • Pipeline
      • 如何工作
      • 细节
    • 参数
    • 机器学习持久化:保存和加载Pipelines
      • 机器学习持久化的向后兼容性
  • 示例代码:
    • 例子:预测器、转换器和参数
    • 例子:Pipeline
    • 模型选择(超参数调试)

Pipelines中的主要概念

MLlib中机器学习算法相关的标准API使得其很容易组合多个算法到一个pipeline或者工作流中,这一部分包括通过Pipelines API介绍的主要概念,以及是从sklearn的哪部分获取的灵感;

  • DataFrame:这个ML API使用Spark SQL中的DataFrame作为ML数据集来持有某一种数据类型,比如一个DataFrame可以有不同类型的列:文本、向量特征、标签和预测结果等;
  • Transformer:转换器是一个可以将某个DataFrame转换成另一个DataFrame的算法,比如一个ML模型就是一个将DataFrame转换为原DataFrame+一个预测列的新的DataFrame的转换器;
  • Estimator:预测器是一个可以fit一个DataFrame得到一个转换器的算法,比如一个学习算法是一个使用DataFrame并训练得到一个模型的预测器;
  • Pipeline:一个Pipeline链使用多个转换器和预测器来指定一个机器学习工作流;
  • Parameter:所有的转换器和预测器通过一个通用API来指定其参数;

DataFrame

机器学习可以作用于很多不同的数据类型,比如向量、文本、图像和结构化数据等,DataFrame属于Spark SQL,支持多种数据类型;

DataFrame支持多种基础和结构化数据;

一个DataFrame可以通过RDD创建;

DataFrame中的列表示名称,比如姓名、年龄、收入等;

Pipeline组件

Transformers - 转换器

转换器是包含特征转换器和学习模型的抽象概念,严格地说,转换器需要实现transform方法,该方法将一个DataFrame转换为另一个DataFrame,通常这种转换是通过在原基础上增加一列或者多列,例如:

  • 一个特征转换器接收一个DataFrame,读取其中一列(比如text),将其映射到一个新的列上(比如feature vector),然后输出一个新的DataFrame包含映射得到的新列;
  • 一个学习模型接收一个DataFrame,读取包含特征向量的列,为每个特征向量预测其标签值,然后输出一个新的DataFrame包含标签列;

Estimators - 预测器

一个预测器是一个学习算法或者任何在数据上使用fit和train的算法的抽象概念,严格地说,一个预测器需要实现fit方法,该方法接收一个DataFrame并产生一个模型,该模型实际上就是一个转换器,例如,逻辑回归是一个预测器,调用其fit方法可以得到一个逻辑回归模型,同时该模型也是一个转换器;

Pipeline组件属性

转换器的transform和预测器的fit都是无状态的,未来可能通过其他方式支持有状态的算法;

每个转换器或者预测器的实例都有一个唯一ID,这在指定参数中很有用;

Pipeline

在机器学习中,运行一系列的算法来处理数据并从数据中学习是很常见的,比如一个简单的文档处理工作流可能包含以下几个步骤:

  • 将每个文档文本切分为单词集合;
  • 将每个文档的单词集合转换为数值特征向量;
  • 使用特征向量和标签学习一个预测模型;

MLlib提供了工作流作为Pipeline,包含一系列的PipelineStageS(转换器和预测器)在指定顺序下运行,我们将使用这个简单工作流作为这一部分的例子;

如何工作

一个Pipeline作为一个特定的阶段序列,每一阶段都是一个转换器或者预测器,这些阶段按顺序执行,输入的DataFrame在每一阶段中都被转换,对于转换器阶段,transform方法作用于DataFrame,对于预测器阶段,fit方法被调用并产生一个转换器(这个转换器会成功Pipeline模型的一部分或者fit pipeline),该转换器的transform方法同样作用于DataFrame上;

下图是一个使用Pipeline的简单文档处理工作流:

Spark Pipeline官方文档

上图中,上面一行表示一个包含三个阶段的Pipeline,Tokenizer和HashingTF为转换器(蓝色),LogisticRegression为预测器(红色),下面一行表示数据流经过整个Pipeline,圆柱体表示DataFrame,Pipeline的fit方法作用于包含原始文本数据和标签的DataFrame,Tokenizer的transform方法将原始文本文档分割为单词集合,作为新列加入到DataFrame中,HashingTF的transform方法将单词集合列转换为特征向量,同样作为新列加入到DataFrame中,目前,LogisticRegression是一个预测器,Pipeline首先调用其fit方法得到一个LogisticRegressionModel,如果Pipeline中还有更多预测器,那么就会在进入下一个阶段前先调用LogisticRegressionModel的transform方法(此时该model就是一个转换器);

一个Pipeline就是一个预测器,因此,在Pipeline的fit方法运行后会产生一个PipelineModel,同样是一个转换器,这个PipelineModel在测试时间使用,下图介绍了该阶段:

Spark Pipeline官方文档

上图中,PipelineModel与原Pipeline有同样数量的阶段,但是原Pipeline中所有的预测器都变成了转换器,当PipelineModel的tranform方法在测试集上调用时,数据将按顺序经过被fit的Pipeline,每个阶段的transform方法将更新DataFrame并传递给下一个阶段;

Pipeline和PipelineModel帮助确定训练和测试数据经过完全一致的特征处理步骤;

细节

DAG Pipeline(有向无环图Pipeline):一个Pipeline的各个阶段被指定作为一个顺序数组,之前的例子都是线性的Pipeline,即每个阶段使用的数据都是前一个阶段提供的,只要数据流图来自于DAG,那么是有可能创建非线性的Pipeline的,这个图是当前指定的基于每个阶段的输入输出列名(通常作为参数指定),如果Pipeline来自DAG,那么各个阶段必须符合拓扑结构顺序;

运行时检查:由于Pipeline可以操作DataFrame可变数据类型,因此它不能使用编译期类型检查,Pipeline和PipelineModel在真正运行会进行运行时检查,这种类型的检查使用DataFrame的schema,schema是一种对DataFrmae中所有数据列数据类型的描述;

唯一Pipeline阶段:一个Pipeline阶段需要是唯一的实例,比如同一个实例myHashingTF不能两次添加到Pipeline中,因为每个阶段必须具备唯一ID,然而,不同的类的实例可以添加到同一个Pipeline中,比如myHashingTF1和myHashingTF2,因为这两个对象有不同的ID,这里的ID可以理解为对象的内容地址,所以myHashingTF2=myHashingTF1也是不行的哈;

参数

MLlib预测器和转换器使用统一API指定参数;

一个参数是各个转换器和预测器自己文档中命名的参数,一个参数Map就是参数的k,v对集合;

这里有两种主要的给算法传参的方式:

  1. 为一个实例设置参数,比如如果lr是逻辑回归的实例对象,可以通过调用lr.setMaxIter(10)指定lr.fit()最多迭代10次,这个API与spark.mllib包中的类似;
  2. 传一个参数Map给fit和transform方法,参数Map中的任何一个参数都会覆盖之前通过setter方法指定的参数;

参数属于转换器和预测器的具体实例,例如,如果我们有两个逻辑回归实例lr1和lr2,然后我们创建一个参数Map,分别指定两个实例的maxIter参数,将会在Pipeline中产生两个参数不同的逻辑回归算法;

机器学习持久化:保存和加载Pipeline

大多数时候为了之后使用将模型或者pipeline持久化到硬盘上是值得的,在Spark 1.6,一个模型的导入/导出功能被添加到了Pipeline的API中,截至Spark 2.3,基于DataFrame的API覆盖了spark.ml和pyspark.ml;

机器学习持久化支持Scala、Java和Python,然而R目前使用一个修改后的格式,因此R存储的模型只能被R加载,这个问题将在未来被修复;

机器学习持久化的向后兼容性

通常来说,MLlib为持久化保持了向后兼容性,即如果你使用某个Spark版本存储了一个模型或者Pipeline,那么你就应该可以通过更新的版本加载它,然而依然有小概率出现异常;

模型持久话:模型或者Pipeline是否通过Spark的X版本存储模型,通过Spark的Y版本加载模型?

  • 主版本:不保证兼容,但是会尽最大努力保持兼容;
  • 次版本和patch版本:保证向后兼容性;
  • 格式提示:不保证有一个稳定的持久化格式,但是模型加载是通过向后兼容性决定的;

模型行为:模型或Pipeline是否在Spark的X版本和Y版本具有一致的行为?

  • 主版本:不保证,但是会尽最大努力保证一致;
  • 次版本和patch版本:行为一致,除非是为了修复bug;

为了模型持久化和模型行为,任何破坏兼容性和一致性的次版本或者patch都会在版本更新笔记中报告出来,如果一个改变没有被报告,那么它应该是为了修复bug出现的;

示例代码

这部分针对上述讨论的内容给出代码示例,更多相关信息,可以查看API文档(ScalaJavaPython);

例子:预测器、转换器和参数

这个例子包含预测器、转换器和参数的主要概念;

Scala:

import org.apache.spark.ml.classification.LogisticRegression
import org.apache.spark.ml.linalg.{Vector, Vectors}
import org.apache.spark.ml.param.ParamMap
import org.apache.spark.sql.Row

// Prepare training data from a list of (label, features) tuples.
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")

// Create a LogisticRegression instance. This instance is an Estimator.
val lr = new LogisticRegression()
// Print out the parameters, documentation, and any default values.
println(s"LogisticRegression parameters:\n ${lr.explainParams()}\n")

// We may set parameters using setter methods.
lr.setMaxIter(10)
  .setRegParam(0.01)

// Learn a LogisticRegression model. This uses the parameters stored in lr.
val model1 = lr.fit(training)
// Since model1 is a Model (i.e., a Transformer produced by an Estimator),
// we can view the parameters it used during fit().
// This prints the parameter (name: value) pairs, where names are unique IDs for this
// LogisticRegression instance.
println(s"Model 1 was fit using parameters: ${model1.parent.extractParamMap}")

// We may alternatively specify parameters using a ParamMap,
// which supports several methods for specifying parameters.
val paramMap = ParamMap(lr.maxIter -> 20)
  .put(lr.maxIter, 30)  // Specify 1 Param. This overwrites the original maxIter.
  .put(lr.regParam -> 0.1, lr.threshold -> 0.55)  // Specify multiple Params.

// One can also combine ParamMaps.
val paramMap2 = ParamMap(lr.probabilityCol -> "myProbability")  // Change output column name.
val paramMapCombined = paramMap ++ paramMap2

// Now learn a new model using the paramMapCombined parameters.
// paramMapCombined overrides all parameters set earlier via lr.set* methods.
val model2 = lr.fit(training, paramMapCombined)
println(s"Model 2 was fit using parameters: ${model2.parent.extractParamMap}")

// Prepare test data.
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")

// Make predictions on test data using the Transformer.transform() method.
// LogisticRegression.transform will only use the 'features' column.
// Note that model2.transform() outputs a 'myProbability' column instead of the usual
// 'probability' column since we renamed the lr.probabilityCol parameter previously.
model2.transform(test)
  .select("features", "label", "myProbability", "prediction")
  .collect()
  .foreach { case Row(features: Vector, label: Double, prob: Vector, prediction: Double) =>
    println(s"($features, $label) -> prob=$prob, prediction=$prediction")
  }

Java:

import java.util.Arrays;
import java.util.List;

import org.apache.spark.ml.classification.LogisticRegression;
import org.apache.spark.ml.classification.LogisticRegressionModel;
import org.apache.spark.ml.linalg.VectorUDT;
import org.apache.spark.ml.linalg.Vectors;
import org.apache.spark.ml.param.ParamMap;
import org.apache.spark.sql.Dataset;
import org.apache.spark.sql.Row;
import org.apache.spark.sql.RowFactory;
import org.apache.spark.sql.types.DataTypes;
import org.apache.spark.sql.types.Metadata;
import org.apache.spark.sql.types.StructField;
import org.apache.spark.sql.types.StructType;

// Prepare training data.
List<Row> dataTraining = Arrays.asList(
    RowFactory.create(1.0, Vectors.dense(0.0, 1.1, 0.1)),
    RowFactory.create(0.0, Vectors.dense(2.0, 1.0, -1.0)),
    RowFactory.create(0.0, Vectors.dense(2.0, 1.3, 1.0)),
    RowFactory.create(1.0, Vectors.dense(0.0, 1.2, -0.5))
);
StructType schema = new StructType(new StructField[]{
    new StructField("label", DataTypes.DoubleType, false, Metadata.empty()),
    new StructField("features", new VectorUDT(), false, Metadata.empty())
});
Dataset<Row> training = spark.createDataFrame(dataTraining, schema);

// Create a LogisticRegression instance. This instance is an Estimator.
LogisticRegression lr = new LogisticRegression();
// Print out the parameters, documentation, and any default values.
System.out.println("LogisticRegression parameters:\n" + lr.explainParams() + "\n");

// We may set parameters using setter methods.
lr.setMaxIter(10).setRegParam(0.01);

// Learn a LogisticRegression model. This uses the parameters stored in lr.
LogisticRegressionModel model1 = lr.fit(training);
// Since model1 is a Model (i.e., a Transformer produced by an Estimator),
// we can view the parameters it used during fit().
// This prints the parameter (name: value) pairs, where names are unique IDs for this
// LogisticRegression instance.
System.out.println("Model 1 was fit using parameters: " + model1.parent().extractParamMap());

// We may alternatively specify parameters using a ParamMap.
ParamMap paramMap = new ParamMap()
  .put(lr.maxIter().w(20))  // Specify 1 Param.
  .put(lr.maxIter(), 30)  // This overwrites the original maxIter.
  .put(lr.regParam().w(0.1), lr.threshold().w(0.55));  // Specify multiple Params.

// One can also combine ParamMaps.
ParamMap paramMap2 = new ParamMap()
  .put(lr.probabilityCol().w("myProbability"));  // Change output column name
ParamMap paramMapCombined = paramMap.$plus$plus(paramMap2);

// Now learn a new model using the paramMapCombined parameters.
// paramMapCombined overrides all parameters set earlier via lr.set* methods.
LogisticRegressionModel model2 = lr.fit(training, paramMapCombined);
System.out.println("Model 2 was fit using parameters: " + model2.parent().extractParamMap());

// Prepare test documents.
List<Row> dataTest = Arrays.asList(
    RowFactory.create(1.0, Vectors.dense(-1.0, 1.5, 1.3)),
    RowFactory.create(0.0, Vectors.dense(3.0, 2.0, -0.1)),
    RowFactory.create(1.0, Vectors.dense(0.0, 2.2, -1.5))
);
Dataset<Row> test = spark.createDataFrame(dataTest, schema);

// Make predictions on test documents using the Transformer.transform() method.
// LogisticRegression.transform will only use the 'features' column.
// Note that model2.transform() outputs a 'myProbability' column instead of the usual
// 'probability' column since we renamed the lr.probabilityCol parameter previously.
Dataset<Row> results = model2.transform(test);
Dataset<Row> rows = results.select("features", "label", "myProbability", "prediction");
for (Row r: rows.collectAsList()) {
  System.out.println("(" + r.get(0) + ", " + r.get(1) + ") -> prob=" + r.get(2)
    + ", prediction=" + r.get(3));
}

Python:

from pyspark.ml.linalg import Vectors
from pyspark.ml.classification import LogisticRegression

# Prepare training data from a list of (label, features) tuples.
training = spark.createDataFrame([
    (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]))], ["label", "features"])

# Create a LogisticRegression instance. This instance is an Estimator.
lr = LogisticRegression(maxIter=10, regParam=0.01)
# Print out the parameters, documentation, and any default values.
print("LogisticRegression parameters:\n" + lr.explainParams() + "\n")

# Learn a LogisticRegression model. This uses the parameters stored in lr.
model1 = lr.fit(training)

# Since model1 is a Model (i.e., a transformer produced by an Estimator),
# we can view the parameters it used during fit().
# This prints the parameter (name: value) pairs, where names are unique IDs for this
# LogisticRegression instance.
print("Model 1 was fit using parameters: ")
print(model1.extractParamMap())

# We may alternatively specify parameters using a Python dictionary as a paramMap
paramMap = {lr.maxIter: 20}
paramMap[lr.maxIter] = 30  # Specify 1 Param, overwriting the original maxIter.
paramMap.update({lr.regParam: 0.1, lr.threshold: 0.55})  # Specify multiple Params.

# You can combine paramMaps, which are python dictionaries.
paramMap2 = {lr.probabilityCol: "myProbability"}  # Change output column name
paramMapCombined = paramMap.copy()
paramMapCombined.update(paramMap2)

# Now learn a new model using the paramMapCombined parameters.
# paramMapCombined overrides all parameters set earlier via lr.set* methods.
model2 = lr.fit(training, paramMapCombined)
print("Model 2 was fit using parameters: ")
print(model2.extractParamMap())

# Prepare test data
test = spark.createDataFrame([
    (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]))], ["label", "features"])

# Make predictions on test data using the Transformer.transform() method.
# LogisticRegression.transform will only use the 'features' column.
# Note that model2.transform() outputs a "myProbability" column instead of the usual
# 'probability' column since we renamed the lr.probabilityCol parameter previously.
prediction = model2.transform(test)
result = prediction.select("features", "label", "myProbability", "prediction") \
    .collect()

for row in result:
    print("features=%s, label=%s -> prob=%s, prediction=%s"
          % (row.features, row.label, row.myProbability, row.prediction))

例子:Pipeline

这个例子是基于上述的简单文本文档处理的例子;

Scala:

import org.apache.spark.ml.{Pipeline, PipelineModel}
import org.apache.spark.ml.classification.LogisticRegression
import org.apache.spark.ml.feature.{HashingTF, Tokenizer}
import org.apache.spark.ml.linalg.Vector
import org.apache.spark.sql.Row

// Prepare training documents from a list of (id, text, label) tuples.
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")

// Configure an ML pipeline, which consists of three stages: tokenizer, hashingTF, and 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))

// Fit the pipeline to training documents.
val model = pipeline.fit(training)

// Now we can optionally save the fitted pipeline to disk
model.write.overwrite().save("/tmp/spark-logistic-regression-model")

// We can also save this unfit pipeline to disk
pipeline.write.overwrite().save("/tmp/unfit-lr-model")

// And load it back in during production
val sameModel = PipelineModel.load("/tmp/spark-logistic-regression-model")

// Prepare test documents, which are unlabeled (id, text) tuples.
val test = spark.createDataFrame(Seq(
  (4L, "spark i j k"),
  (5L, "l m n"),
  (6L, "spark hadoop spark"),
  (7L, "apache hadoop")
)).toDF("id", "text")

// Make predictions on test documents.
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")
  }

Java:

import java.util.Arrays;

import org.apache.spark.ml.Pipeline;
import org.apache.spark.ml.PipelineModel;
import org.apache.spark.ml.PipelineStage;
import org.apache.spark.ml.classification.LogisticRegression;
import org.apache.spark.ml.feature.HashingTF;
import org.apache.spark.ml.feature.Tokenizer;
import org.apache.spark.sql.Dataset;
import org.apache.spark.sql.Row;

// Prepare training documents, which are labeled.
Dataset<Row> training = spark.createDataFrame(Arrays.asList(
  new JavaLabeledDocument(0L, "a b c d e spark", 1.0),
  new JavaLabeledDocument(1L, "b d", 0.0),
  new JavaLabeledDocument(2L, "spark f g h", 1.0),
  new JavaLabeledDocument(3L, "hadoop mapreduce", 0.0)
), JavaLabeledDocument.class);

// Configure an ML pipeline, which consists of three stages: tokenizer, hashingTF, and lr.
Tokenizer tokenizer = new Tokenizer()
  .setInputCol("text")
  .setOutputCol("words");
HashingTF hashingTF = new HashingTF()
  .setNumFeatures(1000)
  .setInputCol(tokenizer.getOutputCol())
  .setOutputCol("features");
LogisticRegression lr = new LogisticRegression()
  .setMaxIter(10)
  .setRegParam(0.001);
Pipeline pipeline = new Pipeline()
  .setStages(new PipelineStage[] {tokenizer, hashingTF, lr});

// Fit the pipeline to training documents.
PipelineModel model = pipeline.fit(training);

// Prepare test documents, which are unlabeled.
Dataset<Row> test = spark.createDataFrame(Arrays.asList(
  new JavaDocument(4L, "spark i j k"),
  new JavaDocument(5L, "l m n"),
  new JavaDocument(6L, "spark hadoop spark"),
  new JavaDocument(7L, "apache hadoop")
), JavaDocument.class);

// Make predictions on test documents.
Dataset<Row> predictions = model.transform(test);
for (Row r : predictions.select("id", "text", "probability", "prediction").collectAsList()) {
  System.out.println("(" + r.get(0) + ", " + r.get(1) + ") --> prob=" + r.get(2)
    + ", prediction=" + r.get(3));
}

Python:

from pyspark.ml import Pipeline
from pyspark.ml.classification import LogisticRegression
from pyspark.ml.feature import HashingTF, Tokenizer

# Prepare training documents from a list of (id, text, label) tuples.
training = spark.createDataFrame([
    (0, "a b c d e spark", 1.0),
    (1, "b d", 0.0),
    (2, "spark f g h", 1.0),
    (3, "hadoop mapreduce", 0.0)
], ["id", "text", "label"])

# Configure an ML pipeline, which consists of three stages: tokenizer, hashingTF, and lr.
tokenizer = Tokenizer(inputCol="text", outputCol="words")
hashingTF = HashingTF(inputCol=tokenizer.getOutputCol(), outputCol="features")
lr = LogisticRegression(maxIter=10, regParam=0.001)
pipeline = Pipeline(stages=[tokenizer, hashingTF, lr])

# Fit the pipeline to training documents.
model = pipeline.fit(training)

# Prepare test documents, which are unlabeled (id, text) tuples.
test = spark.createDataFrame([
    (4, "spark i j k"),
    (5, "l m n"),
    (6, "spark hadoop spark"),
    (7, "apache hadoop")
], ["id", "text"])

# Make predictions on test documents and print columns of interest.
prediction = model.transform(test)
selected = prediction.select("id", "text", "probability", "prediction")
for row in selected.collect():
    rid, text, prob, prediction = row
    print("(%d, %s) --> prob=%s, prediction=%f" % (rid, text, str(prob), prediction))

模型选择(超参数调试)

机器学习Pipeline的一个巨大用处是调参,点击这里获取更多自动模型选择的相关信息;

上一篇:LoadRunner封装函数


下一篇:python新手学习记录图像处理2,镜像(上下镜像与左右镜像)