您的位置:首页 > 大数据

图解大数据 | Spark机器学习(上)-工作流与特征工程

2022-03-08 23:37 1541 查看

作者:韩信子@ShowMeAI 教程地址http://www.showmeai.tech/tutorials/84 本文地址http://www.showmeai.tech/article-detail/180 声明:版权所有,转载请联系平台与作者并注明出处

1.Spark机器学习工作流

1)Spark mllib 与ml

Spark中同样有用于大数据机器学习的板块MLlib/ML,可以支持对海量数据进行建模与应用。

2)机器学习工作流(Pipeline)

一个典型的机器学习过程,从数据收集开始,要经历多个步骤,才能得到需要的输出。是一个包含多个步骤的流水线式工作:

  • 源数据ETL(抽取、转化、加载)
  • 数据预处理
  • 指标提取
  • 模型训练与交叉验证
  • 新数据预测

MLlib 已足够简单易用,但在一些情况下使用 MLlib 将会让程序结构复杂,难以理解和实现。

  • 目标数据集结构复杂需要多次处理。
  • 对新数据进行预测的时候,需要结合多个已经训练好的单个模型进行综合预测 Spark 1.2 版本之后引入的 ML Pipeline,可以用于构建复杂机器学习工作流应用。

以下是几个重要概念的解释:

(1)DataFrame

使用Spark SQL中的 DataFrame 作为数据集,可以容纳各种数据类型。较之 RDD,DataFrame 包含了 schema 信息,更类似传统数据库中的二维表格。

它被 ML Pipeline 用来存储源数据,例如DataFrame 中的列可以是存储的文本、特征向量、真实标签和预测的标签等。

(2)Transformer(转换器)

是一种可以将一个DataFrame 转换为另一个DataFrame 的算法。比如,一个模型就是一个 Transformer,它可以把一个不包含预测标签的测试数据集 DataFrame 打上标签,转化成另一个包含预测标签的 DataFrame。

技术上,Transformer实现了一个方法transform(),通过附加一个或多个列将一个 DataFrame 转换为另一个DataFrame。

(3)Estimator(估计器/评估器)

是学习算法或在训练数据上的训练方法的概念抽象。在 Pipeline 里通常是被用来操作 DataFrame 数据,并生产一个 Transformer。从技术上讲,Estimator 实现了一个方法fit(),它接受一个DataFrame 并产生一个Transformer转换器。

(4)Parameter

Parameter 被用来设置 Transformer 或者 Estimator 的参数。现在,所有 Transformer(转换器)和Estimator(估计器)可共享用于指定参数的公共API。ParamMap是一组(参数,值)对。

(5)PipeLine(工作流/管道)

工作流将多个工作流阶段( Transformer转换器和Estimator估计器)连接在一起,形成机器学习的工作流,并获得结果输出。

3)构建一个Pipeline工作流

val pipeline = new Pipeline().setStages(Array(stage1,stage2,stage3,…))

① 首先需要定义 Pipeline 中的各个PipelineStage(工作流阶段)

  • 包括Transformer转换器 和Estimator评估器。
  • 比如指标提取 和 转换模型训练。
  • 有了这些处理特定问题的Transformer转换器和 Estimator评估器,就可以按照具体的处理逻辑,有序地组织PipelineStages,并创建一个Pipeline。

② 然后,可以把训练数据集作为入参,并调用 Pipelin 实例的 fit 方法,开始以流的方式来处理源训练数据

  • 这个调用会返回一个 PipelineModel 类实例,进而被用来预测测试数据的标签

③ 工作流的各个阶段按顺序运行,输入的DataFrame在它通过每个阶段时被转换

  • 对于 Transformer转换器阶段,在DataFrame上调用
    transform()
    方法。
  • 对于Estimator估计器阶段,调用fit()方法来生成一个转换器(它成为PipelineModel的一部分或拟合的Pipeline),并且在DataFrame上调用该转换器的
    transform()
    方法。

4)构建Pipeline示例

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  # type: ignore
print(
"(%d, %s) --> prob=%s, prediction=%f" % (
rid, text, str(prob), prediction   # type: ignore
)
)

2.基于DataFrame的Spark ML特征工程

1)特征工程

2)二值化

continuousDataFrame = spark.createDataFrame([(0, 1.1),(1, 8.5),(2, 5.2)], ["id", "feature"])
binarizer = Binarizer(threshold=5.1, inputCol="feature", outputCol="binarized_feature")
binarizedDataFrame = binarizer.transform(continuousDataFrame)

3)定边界离散化

splits = [-float("inf"), -0.5, 0.0, 0.5, float("inf")]
data = [(-999.9,),(-0.5,),(-0.3,),(0.0,),(0.2,),(999.9,)]
dataFrame = spark.createDataFrame(data, ["features"])
bucketizer = Bucketizer(splits=splits, inputCol="features", outputCol="bucketedFeatures")

# 按照给定的边界进行分桶
bucketedData = bucketizer.transform(dataFrame)

4)按照分位数离散化

data = [(0, 18.0), (1, 19.0), (2, 8.0), (3, 5.0), (4, 2.2), (5, 9.2), (6, 14.4)]
df = spark.createDataFrame(data, ["id", "hour"])
df = df.repartition(1)

# 分成3个桶进行离散化
discretizer = QuantileDiscretizer(numBuckets=3, inputCol="hour", outputCol="result")
result = discretizer.fit(df).transform(df)

5)连续值幅度缩放

dataFrame = spark.createDataFrame([
(0, Vectors.dense([1.0, 0.1, -8.0]),),
(1, Vectors.dense([2.0, 1.0, -4.0]),),
(2, Vectors.dense([4.0, 10.0, 8.0]),)
], ["id", "features"])
scaler = MaxAbsScaler(inputCol="features", outputCol="scaledFeatures")

# 计算最大绝对值用于缩放
scalerModel = scaler.fit(dataFrame)

# 缩放幅度到[-1, 1]之间
scaledData = scalerModel.transform(dataFrame)

6)标准化

dataFrame = spark.read.format("libsvm").load("data/mllib/sample_libsvm_data.txt")
scaler = StandardScaler(inputCol="features", outputCol="scaledFeatures", withStd=True, withMean=False)

# 计算均值方差等参数
scalerModel = scaler.fit(dataFrame)

# 标准化
scaledData = scalerModel.transform(dataFrame)

7)添加多项式特征

df = spark.createDataFrame([(Vectors.dense([2.0, 1.0]),), (Vectors.dense([0.0, 0.0]),), (Vectors.dense([3.0, -1.0]),)], ["features"])
polyExpansion = PolynomialExpansion(degree=3, inputCol="features", outputCol="polyFeatures")
polyDF = polyExpansion.transform(df)

8)类别型独热向量编码

df = spark.createDataFrame([ (0,"a"), (1,"b"), (2,"c"), (3,"a"), (4,"a"), (5,"c")], ["id","category"])
stringIndexer = StringIndexer(inputCol="category", outputCol="categoryIndex")
model = stringIndexer.fit(df)
indexed = model.transform(df)

encoder = OneHotEncoder(inputCol="categoryIndex", outputCol="categoryVec")
encoded = encoder.transform(indexed)

9)文本型特征抽取

df = spark.createDataFrame([(0, "a b c".split(" ")), (1, "a b b c a".split(" "))], ["id", "words"])
cv = CountVectorizer(inputCol="words", outputCol="features", vocabSize=3, minDF=2.0)
model = cv.fit(df)
result = model.transform(df)

10)文本型特征抽取

sentenceData = spark.createDataFrame([
(0.0, "Hi I heard about Spark"),
(0.0, "I wish Java could use case classes"),
(1.0, "Logistic regression models are neat")
], ["label", "sentence"])

tokenizer = Tokenizer(inputCol="sentence", outputCol="words")
wordsData = tokenizer.transform(sentenceData)
hashingTF = HashingTF(inputCol="words", outputCol="rawFeatures", numFeatures=20)
featurizedData = hashingTF.transform(wordsData)
idf = IDF(inputCol="rawFeatures", outputCol="features")
idfModel = idf.fit(featurizedData)
rescaledData = idfModel.transform(featurizedData)

3.参考资料

ShowMeAI相关文章推荐

ShowMeAI系列教程推荐

内容来自用户分享和网络整理,不保证内容的准确性,如有侵权内容,可联系管理员处理 点击这里给我发消息
标签: