数据挖掘特征工程方法总结 (PySpark)

• Normalizer

• StandardScaler

• RobustScaler

• MinMaxScaler

• MaxAbsScaler

• Bucketizer

• ElementwiseProduct

• SQLTransformer

• VectorAssembler

• QuantileDiscretizer

• Imputer

Normalizer

Normalizer 是一个 Transformer ，它转换一个包含Vector行的数据集，将每个Vector规范化为单位范数。它接受参数p，它指定用于正常化的p-norm。默认(p = 2)。

from pyspark.ml.feature import Normalizerfrom pyspark.ml.linalg import VectorsdataFrame = spark.createDataFrame([    (0, Vectors.dense([1.0, 0.5, -1.0]),),    (1, Vectors.dense([2.0, 1.0, 1.0]),),    (2, Vectors.dense([4.0, 10.0, 2.0]),)], ["id", "features"])# 使用$L^1$范数对每个向量进行规范化。normalizer = Normalizer(inputCol="features",                         outputCol="normFeatures", p=1.0)l1NormData = normalizer.transform(dataFrame)print("Normalized using L^1 norm")l1NormData.show()# 使用 $L^\infty$ 范数对每个向量进行规范化。lInfNormData = normalizer.transform(dataFrame, {normalizer.p: float("inf")})print("Normalized using L^inf norm")lInfNormData.show()

Normalized using L^1 norm+---+--------------+------------------+| id|      features|      normFeatures|+---+--------------+------------------+|  0|[1.0,0.5,-1.0]|    [0.4,0.2,-0.4]||  1| [2.0,1.0,1.0]|   [0.5,0.25,0.25]||  2|[4.0,10.0,2.0]|[0.25,0.625,0.125]|+---+--------------+------------------+Normalized using L^inf norm+---+--------------+--------------+| id|      features|  normFeatures|+---+--------------+--------------+|  0|[1.0,0.5,-1.0]|[1.0,0.5,-1.0]||  1| [2.0,1.0,1.0]| [1.0,0.5,0.5]||  2|[4.0,10.0,2.0]| [0.4,1.0,0.2]|+---+--------------+--------------+

StandardScaler

StandardScaler转换一个数据集的向量行，规范化每个特征标准差为1和/均值为0。它需要参数:

• withStd:默认为True。将数据缩放到单位标准差。
• withMean:默认为False。在缩放前将数据以平均值居中。它将构建一个密集的输出，所以在应用于稀疏输入时要小心。

from pyspark.ml.feature import StandardScalerdataFrame = spark.read.format("libsvm").load("sample_libsvm_data.txt")scaler = StandardScaler(inputCol="features", outputCol="scaledFeatures",                        withStd=True, withMean=False)# 通过拟合StandardScaler计算汇总统计信息scalerModel = scaler.fit(dataFrame)# 将每个特征归一化，使其具有单位标准差。scaledData = scalerModel.transform(dataFrame)scaledData.show(5)

+-----+--------------------+--------------------+|label|            features|      scaledFeatures|+-----+--------------------+--------------------+|  0.0|(692,[127,128,129...|(692,[127,128,129...||  1.0|(692,[158,159,160...|(692,[158,159,160...||  1.0|(692,[124,125,126...|(692,[124,125,126...||  1.0|(692,[152,153,154...|(692,[152,153,154...||  1.0|(692,[151,152,153...|(692,[151,152,153...|+-----+--------------------+--------------------+only showing top 5 rows

RobustScaler

RobustScaler转换由Vector行组成的数据集，删除中值并根据特定的分位数范围缩放数据(默认为IQR: Interquartile range，第1和第3分位数之间的分位数范围)。它的行为与StandardScaler非常相似，但是使用的是中位数和分位数范围，而不是平均值和标准差，这使得它对异常值具有鲁棒性。它需要参数:

• lower:默认为0.25。下分位数计算分位数范围，由所有特性共享。
• Upper:默认为0.75。上分位数计算分位数范围，由所有特征共享。
• withScaling:默认为True。将数据缩放到分位数范围。
• withcentric:默认为False。在缩放前用中位数将数据居中。

from pyspark.ml.feature import RobustScalerdataFrame = spark.read.format("libsvm").load("data/mllib/sample_libsvm_data.txt")scaler = RobustScaler(inputCol="features", outputCol="scaledFeatures",                      withScaling=True, withCentering=False,                      lower=0.25, upper=0.75)# 通过拟合RobustScaler计算汇总统计数据scalerModel = scaler.fit(dataFrame)# 将每个特征转换为单位分位数范围。scaledData = scalerModel.transform(dataFrame)scaledData.show(5)

+-----+--------------------+--------------------+|label|            features|      scaledFeatures|+-----+--------------------+--------------------+|  0.0|(692,[127,128,129...|(692,[127,128,129...||  1.0|(692,[158,159,160...|(692,[158,159,160...||  1.0|(692,[124,125,126...|(692,[124,125,126...||  1.0|(692,[152,153,154...|(692,[152,153,154...||  1.0|(692,[151,152,153...|(692,[151,152,153...|+-----+--------------------+--------------------+only showing top 5 rows

MinMaxScaler

MinMaxScaler转换一个Vector行数据集，将每个特性缩放到特定的范围(通常是[0,1])。它需要参数:

• min:默认为0.0。转换后的下界
• Max:默认为1.0。转换后的上界

from pyspark.ml.feature import MinMaxScalerfrom pyspark.ml.linalg import VectorsdataFrame = spark.createDataFrame([    (0, Vectors.dense([1.0, 0.1, -1.0]),),    (1, Vectors.dense([2.0, 1.1, 1.0]),),    (2, Vectors.dense([3.0, 10.1, 3.0]),)], ["id", "features"])scaler = MinMaxScaler(inputCol="features", outputCol="scaledFeatures")# 计算汇总统计并生成MinMaxScalerModelscalerModel = scaler.fit(dataFrame)# 缩放每个特征到 [min, max] 范围内scaledData = scalerModel.transform(dataFrame)print("Features scaled to range: [%f, %f]" % (scaler.getMin(), scaler.getMax()))scaledData.select("features", "scaledFeatures").show()

Features scaled to range: [0.000000, 1.000000]+--------------+--------------+|      features|scaledFeatures|+--------------+--------------+|[1.0,0.1,-1.0]|     (3,[],[])|| [2.0,1.1,1.0]| [0.5,0.1,0.5]||[3.0,10.1,3.0]| [1.0,1.0,1.0]|+--------------+--------------+

MaxAbsScaler

MaxAbsScaler转换一个由Vector行组成的数据集，通过除以每个特征中的最大绝对值将每个特征缩放到[-1,1]范围。它不平移数据，因此不破坏任何稀疏性。

from pyspark.ml.feature import MaxAbsScalerfrom pyspark.ml.linalg import VectorsdataFrame = 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")# 计算汇总统计并生成MaxAbsScalerModelscalerModel = scaler.fit(dataFrame)# 缩放每个特征到 [-1, 1] 范围内scaledData = scalerModel.transform(dataFrame)scaledData.select("features", "scaledFeatures").show()

+--------------+--------------------+|      features|      scaledFeatures|+--------------+--------------------+|[1.0,0.1,-8.0]|[0.25,0.010000000...||[2.0,1.0,-4.0]|      [0.5,0.1,-0.5]||[4.0,10.0,8.0]|       [1.0,1.0,1.0]|+--------------+--------------------+

Bucketizer

分桶器Bucketizer将一列连续的特性转换为一列特性桶，其中桶由用户指定。它接受一个参数:

• splits:将连续特性映射到桶的参数。分成n+1次，就有n个桶。一个由拆分x,y定义的桶，除了最后一个桶(也包含y)，其余的值都在[x,y]范围内。拆分应该严格递增。必须显式提供-inf和inf的值以覆盖所有Double值；否则，在指定的拆分符之外的值将被视为错误。

from pyspark.ml.feature import Bucketizersplits = [-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)print("Bucketizer output with %d buckets" % (len(bucketizer.getSplits()) - 1))bucketedData.show()

Bucketizer output with 4 buckets+--------+----------------+|features|bucketedFeatures|+--------+----------------+|  -999.9|             0.0||    -0.5|             1.0||    -0.3|             1.0||     0.0|             2.0||     0.2|             2.0||   999.9|             3.0|+--------+----------------+

ElementwiseProduct

ElementwiseProduct使用元素级乘法，将每个输入向量乘以一个提供的“权重”向量。换句话说，它将数据集的每一列缩放一个标量乘数。这表示输入向量v和变换向量w之间的阿达玛乘积，从而得到结果向量。

from pyspark.ml.feature import ElementwiseProductfrom pyspark.ml.linalg import Vectors# 创建一些矢量数据;同样适用于稀疏向量data = [(Vectors.dense([1.0, 2.0, 3.0]),),         (Vectors.dense([4.0, 5.0, 6.0]),)]df = spark.createDataFrame(data, ["vector"])transformer = ElementwiseProduct(scalingVec=Vectors.dense([0.0, 1.0, 2.0]),                                 inputCol="vector",                                  outputCol="transformedVector")# 批量转换向量以创建新列:transformer.transform(df).show()

+-------------+-----------------+|       vector|transformedVector|+-------------+-----------------+|[1.0,2.0,3.0]|    [0.0,2.0,6.0]||[4.0,5.0,6.0]|   [0.0,5.0,12.0]|+-------------+-----------------+

SQLTransformer

SQLTransformer实现由SQL语句定义的转换。目前，我们只支持像“SELECT…”FROM __THIS__…"其中"_ THIS _"表示输入数据集的底层表。select子句指定在输出中显示的字段、常量和表达式，可以是Spark SQL支持的任何select子句。用户还可以使用Spark SQL内置函数和udf对这些选中的列进行操作。例如，SQLTransformer支持这样的语句:

• SELECT a, a + b AS a_b FROM __THIS__
• SELECT a, SQRT(b) AS b_sqrt FROM __THIS__ where a > 5
• SELECT a, b, SUM(c) AS c_sum FROM __THIS__ GROUP BY a, b

from pyspark.ml.feature import SQLTransformerdf = spark.createDataFrame([    (0, 1.0, 3.0),    (2, 2.0, 5.0)], ["id", "v1", "v2"])sqlTrans = SQLTransformer(    statement="SELECT *, (v1 + v2) AS v3, (v1 * v2) AS v4 FROM __THIS__")sqlTrans.transform(df).show()

+---+---+---+---+----+| id| v1| v2| v3|  v4|+---+---+---+---+----+|  0|1.0|3.0|4.0| 3.0||  2|2.0|5.0|7.0|10.0|+---+---+---+---+----+

VectorAssembler

VectorAssembler是一个转换器，它将给定的列列表组合成单个向量列。它可以将原始特征和由不同特征转换器生成的特征组合成一个特征向量，以便训练逻辑回归和决策树等ML模型。VectorAssembler接受以下输入列类型：所有数值类型、布尔类型和向量类型。在每一行中，输入列的值将按照指定的顺序连接到一个向量中。

from pyspark.ml.linalg import Vectorsfrom pyspark.ml.feature import VectorAssemblerdataset = spark.createDataFrame(    [(0, 18, 1.0, Vectors.dense([0.0, 10.0, 0.5]), 1.0)],    ["id", "hour", "mobile", "userFeatures", "clicked"])assembler = VectorAssembler(    inputCols=["hour", "mobile", "userFeatures"],    outputCol="features")output = assembler.transform(dataset)print("Assembled columns 'hour', 'mobile', 'userFeatures' to vector column 'features'")output.select("features", "clicked").show(truncate=False)

Assembled columns 'hour', 'mobile', 'userFeatures' to vector column 'features'+-----------------------+-------+|features               |clicked|+-----------------------+-------+|[18.0,1.0,0.0,10.0,0.5]|1.0    |+-----------------------+-------+

QuantileDiscretizer

QuantileDiscretizer 采用具有连续特征的列并输出具有分箱分类特征的列。bin 的数量由 numBuckets 参数设置。使用的桶数可能会小于此值，例如，如果输入的不同值太少而无法创建足够的不同分位数。

NaN 值：NaN 值将在 QuantileDiscretizer 拟合期间从列中删除。这将产生一个 Bucketizer 用于预测的模型。在转换过程中， Bucketizer 当它在数据集中找到 NaN 值时会引发错误，但用户也可以通过设置来选择保留或删除数据集中的 NaN 值 handleInvalid 。如果用户选择保留 NaN 值，会进行特殊处理，放入自己的桶中，例如，如果使用 4 个桶，则非 NaN 数据将放入桶[0-3]，但 NaN 将计数在一个特殊的桶中[4]。

from pyspark.ml.feature import QuantileDiscretizerdata = [(0, 18.0), (1, 19.0), (2, 8.0), (3, 5.0), (4, 2.2)]df = spark.createDataFrame(data, ["id", "hour"])discretizer = QuantileDiscretizer(numBuckets=3,                                   inputCol="hour",                                  outputCol="result")result = discretizer.fit(df).transform(df)result.show()

+---+----+------+| id|hour|result|+---+----+------+|  0|18.0|   2.0||  1|19.0|   2.0||  2| 8.0|   1.0||  3| 5.0|   1.0||  4| 2.2|   0.0|+---+----+------+

Imputer

估计器使用 Imputer 缺失值所在列的平均值、中位数或众数来完成数据集中的缺失值。输入列应该是数字类型。目前 Imputer 不支持分类特征，并且可能为包含分类特征的列创建不正确的值。Imputer 可以通过 'NaN' 估算自定义值 .setMissingValue(custom_value) 。例如， .setMissingValue(0) 将估算所有出现的 (0)。

from pyspark.ml.feature import Imputerdf = spark.createDataFrame([    (1.0, float("nan")),    (2.0, float("nan")),    (float("nan"), 3.0),    (4.0, 4.0),    (5.0, 5.0)], ["a", "b"])imputer = Imputer(inputCols=["a", "b"], outputCols=["out_a", "out_b"])model = imputer.fit(df)model.transform(df).show()

+---+---+-----+-----+|  a|  b|out_a|out_b|+---+---+-----+-----+|1.0|NaN|  1.0|  4.0||2.0|NaN|  2.0|  4.0||NaN|3.0|  3.0|  3.0||4.0|4.0|  4.0|  4.0||5.0|5.0|  5.0|  5.0|+---+---+-----+-----+