曾經在15、16年那會兒使用Spark做機器學習，那時候pyspark并不成熟，做特征工程主要還是寫scala，后來進入阿里作業，特征處理基本上使用PAI 可視化特征工程組件+ODPS SQL，復雜的話才會自己寫python處理，最近重新學習了下pyspark，筆記下如何使用pyspark做特征工程，

我們使用movielens的資料進行，oneHotEncoder、multiHotEncoder和Numerical features的特征處理，

main

from pyspark import SparkConf
from pyspark.ml import Pipeline
from pyspark.ml.feature import OneHotEncoder, StringIndexer, QuantileDiscretizer, MinMaxScaler
from pyspark.ml.linalg import VectorUDT, Vectors
from pyspark.sql import SparkSession
from pyspark.sql.functions import *
from pyspark.sql.types import *
from pyspark.sql import functions as F

if __name__ == '__main__':
    conf = SparkConf().setAppName('featureEngineering').setMaster('local')
    spark = SparkSession.builder.config(conf=conf).getOrCreate()
    file_path = 'file:///資源檔案夾路徑'
    movieResourcesPath = file_path + "/webroot/sampledata/movies.csv"
    movieSamples = spark.read.format('csv').option('header', 'true').load(movieResourcesPath)
    print("Raw Movie Samples:")
    movieSamples.show(10)
    movieSamples.printSchema()
    print("OneHotEncoder Example:")
    oneHotEncoderExample(movieSamples)
    print("MultiHotEncoder Example:")
    multiHotEncoderExample(movieSamples)
    print("Numerical features Example:")
    ratingsResourcesPath = file_path + "/webroot/sampledata/ratings.csv"
    ratingSamples = spark.read.format('csv').option('header', 'true').load(ratingsResourcesPath)
    ratingFeatures(ratingSamples)

我們先來看看“movies.csv” 和 “ratings.csv” 資料長什么樣子吧：

movies samples:
+-------+--------------------+--------------------+
|movieId|               title|              genres|
+-------+--------------------+--------------------+
|      1|    Toy Story (1995)|Adventure|Animati...|
|      2|      Jumanji (1995)|Adventure|Childre...|
|      3|Grumpier Old Men ...|      Comedy|Romance|
|      4|Waiting to Exhale...|Comedy|Drama|Romance|
|      5|Father of the Bri...|              Comedy|
+-------+--------------------+--------------------+

ratings samples:
+------+-------+------+----------+
|userId|movieId|rating| timestamp|
+------+-------+------+----------+
|     1|      2|   3.5|1112486027|
|     1|     29|   3.5|1112484676|
|     1|     32|   3.5|1112484819|
|     1|     47|   3.5|1112484727|
|     1|     50|   3.5|1112484580|
+------+-------+------+----------+

oneHotEncoder

我們對movieId進行oneHotEncoder：

def oneHotEncoderExample(movieSamples):
    # 把movieId的值，轉為int直接作為movieIdNumber編號
    samplesWithIdNumber = movieSamples.withColumn("movieIdNumber", F.col("movieId").cast(IntegerType()))
    encoder = OneHotEncoder(inputCols=["movieIdNumber"], outputCols=['movieIdVector'], dropLast=False)
    oneHotEncoderSamples = encoder.fit(samplesWithIdNumber).transform(samplesWithIdNumber)
    oneHotEncoderSamples.printSchema()
    oneHotEncoderSamples.show(5)

OneHotEncoder Example:
root
 |-- movieId: string (nullable = true)
 |-- title: string (nullable = true)
 |-- genres: string (nullable = true)
 |-- movieIdNumber: integer (nullable = true)
 |-- movieIdVector: vector (nullable = true)

+-------+--------------------+--------------------+-------------+----------------+
|movieId|               title|              genres|movieIdNumber|   movieIdVector|
+-------+--------------------+--------------------+-------------+----------------+
|      1|    Toy Story (1995)|Adventure|Animati...|            1|(1001,[1],[1.0])|
|      2|      Jumanji (1995)|Adventure|Childre...|            2|(1001,[2],[1.0])|
|      3|Grumpier Old Men ...|      Comedy|Romance|            3|(1001,[3],[1.0])|
|      4|Waiting to Exhale...|Comedy|Drama|Romance|            4|(1001,[4],[1.0])|
|      5|Father of the Bri...|              Comedy|            5|(1001,[5],[1.0])|
+-------+--------------------+--------------------+-------------+----------------+
only showing top 5 rows

multiHotEncoder

我們再對電影型別‘genres’進行multiHotEncoder：

def multiHotEncoderExample(movieSamples):
    # 對genres進行切分，一行變多行
    samplesWithGenre = movieSamples.select("movieId", "title", explode(
        split(F.col("genres"), "\\|").cast(ArrayType(StringType()))).alias('genre'))
    print("samplesWithGenre Samples:")
    samplesWithGenre.printSchema()
    samplesWithGenre.show(5)

samplesWithGenre Samples:
root
 |-- movieId: string (nullable = true)
 |-- title: string (nullable = true)
 |-- genre: string (nullable = true)

+-------+----------------+---------+
|movieId|           title|    genre|
+-------+----------------+---------+
|      1|Toy Story (1995)|Adventure|
|      1|Toy Story (1995)|Animation|
|      1|Toy Story (1995)| Children|
|      1|Toy Story (1995)|   Comedy|
|      1|Toy Story (1995)|  Fantasy|
+-------+----------------+---------+
only showing top 5 rows

    # genre 編碼
    genreIndexer = StringIndexer(inputCol="genre", outputCol="genreIndex")
    StringIndexerModel = genreIndexer.fit(samplesWithGenre)
    genreIndexSamples = StringIndexerModel.transform(samplesWithGenre).withColumn("genreIndexInt",
                                                                                  F.col("genreIndex").cast(IntegerType()))
    # 計算編碼向量大小
    indexSize = genreIndexSamples.agg(max(F.col("genreIndexInt"))).head()[0] + 1
    # 根據 movieId 聚合genreIndexInt
    processedSamples = genreIndexSamples.groupBy('movieId').agg(
        F.collect_list('genreIndexInt').alias('genreIndexes')).withColumn("indexSize", F.lit(indexSize))
    # 生成vector
    finalSample = processedSamples.withColumn("vector",
                                              udf(array2vec, VectorUDT())(F.col("genreIndexes"), F.col("indexSize")))
    print("finalSample Samples:")
    finalSample.printSchema()
    finalSample.show(5)

finalSample Samples:
root
 |-- movieId: string (nullable = true)
 |-- genreIndexes: array (nullable = true)
 |    |-- element: integer (containsNull = false)
 |-- indexSize: integer (nullable = false)
 |-- vector: vector (nullable = true)

+-------+------------+---------+--------------------+
|movieId|genreIndexes|indexSize|              vector|
+-------+------------+---------+--------------------+
|    296|[1, 5, 0, 3]|       19|(19,[0,1,3,5],[1....|
|    467|         [1]|       19|      (19,[1],[1.0])|
|    675|   [4, 0, 3]|       19|(19,[0,3,4],[1.0,...|
|    691|      [1, 2]|       19|(19,[1,2],[1.0,1.0])|
|    829| [1, 10, 14]|       19|(19,[1,10,14],[1....|
+-------+------------+---------+--------------------+
only showing top 5 rows

其中生成vector的udf array2vec ：

def array2vec(genreIndexes, indexSize):
    genreIndexes.sort()
    fill_list = [1.0 for _ in range(len(genreIndexes))]
    # 稀疏向量存盤 indexSize，有值的Indexes，對應Indexes上的填充值
    return Vectors.sparse(indexSize, genreIndexes, fill_list)

Numerical features

對于Numerical features，我們可以進行分桶或者標準化，在這里，先我們讀取“ratings.csv”資料，統計各電影被評價的次數以及平均得分：

def ratingFeatures(ratingSamples):
    # calculate average movie rating score and rating count
    movieFeatures = ratingSamples.groupBy('movieId').agg(F.count(F.lit(1)).alias('ratingCount'),
                                                         F.avg("rating").alias("avgRating"))\
        .withColumn('avgRatingVec', udf(lambda x: Vectors.dense(x), VectorUDT())('avgRating'))
    print("movieFeatures:")
    movieFeatures.show(5)

movieFeatures:
+-------+-----------+------------------+--------------------+
|movieId|ratingCount|         avgRating|        avgRatingVec|
+-------+-----------+------------------+--------------------+
|    296|      14616| 4.165606185002737| [4.165606185002737]|
|    467|        174|3.4367816091954024|[3.4367816091954024]|
|    829|        402|2.6243781094527363|[2.6243781094527363]|
|    691|        254|3.1161417322834644|[3.1161417322834644]|
|    675|          6|2.3333333333333335|[2.3333333333333335]|
+-------+-----------+------------------+--------------------+
only showing top 5 rows

再對被評價的次數進行分桶，對平均得分進行標準化：

    # bucketing
    ratingCountDiscretizer = QuantileDiscretizer(numBuckets=100, inputCol="ratingCount", outputCol="ratingCountBucket")
    # Normalization
    ratingScaler = MinMaxScaler(inputCol="avgRatingVec", outputCol="scaleAvgRating")
    pipelineStage = [ratingCountDiscretizer, ratingScaler]
    featurePipeline = Pipeline(stages=pipelineStage)
    movieProcessedFeatures = featurePipeline.fit(movieFeatures).transform(movieFeatures)
    movieProcessedFeatures.show(5)

+-------+-----------+------------------+--------------------+-----------------+--------------------+
|movieId|ratingCount|         avgRating|        avgRatingVec|ratingCountBucket|      scaleAvgRating|
+-------+-----------+------------------+--------------------+-----------------+--------------------+
|    296|      14616| 4.165606185002737| [4.165606185002737]|             57.0|[0.9170998054196596]|
|    467|        174|3.4367816091954024|[3.4367816091954024]|             21.0|[0.7059538707722662]|
|    829|        402|2.6243781094527363|[2.6243781094527363]|             32.0|[0.4705944962973248]|
|    691|        254|3.1161417322834644|[3.1161417322834644]|             26.0|[0.6130620985364005]|
|    675|          6|2.3333333333333335|[2.3333333333333335]|              3.0|[0.38627664627161...|
+-------+-----------+------------------+--------------------+-----------------+--------------------+
only showing top 5 rows