2025年05月02日玄貓（BlackCat）

資料前處理與特徵工程技術探討

本文探討資料科學專案中缺失值處理與變數選擇的重要性，並介紹 PySpark 中的 StringIndexer、VectorAssembler 和資料壓縮技術 PCA 與 SVD 的應用。文章涵蓋缺失值型別、處理方法、基數與缺失值比例計算，以及程式碼範例和圖表說明，提供實務操作指引。

資料科學機器學習

缺失值處理變數選擇特徵工程 PySpark PCA SVD

在資料科學專案中，資料品質攸關模型成效，因此前處理步驟至關重要。本文首先介紹缺失值處理，包含完全隨機缺失、隨機缺失和非隨機缺失三種型別，並說明均值填補、模型填補、多重填補等處理策略。接著探討變數選擇，根據基數和缺失值比例等指標進行篩選，提供 PySpark 程式碼示範如何計算和應用這些指標。進一步說明特徵工程中類別變數的轉換技巧，使用 StringIndexer 將類別變數轉換為數值型態，並運用 VectorAssembler 組合特徵向量，為機器學習模型訓練做好準備。最後，文章簡述 PCA 和 SVD 等資料壓縮技術，說明其原理和應用場景，以提升模型效率和效能。

缺失值處理與變數選擇

在資料科學專案中，缺失值的處理是至關重要的一環。資料缺失的原因多種多樣，理解缺失值的型別有助於選擇適當的處理方法。

缺失值的型別

完全隨機缺失（MCAR）：資料的缺失與任何變數都無關。這種情況下，缺失值的出現是完全隨機的，不會對資料分析產生系統性的偏差。
範例：假設某個房屋價格資料集中的價格資訊缺失是完全隨機的。在這種情況下，可以使用均值、中位數或眾數填補缺失值。
隨機缺失（MAR）：資料的缺失與觀測到的變數有關，但與未觀測到的變數無關。也就是說，缺失值的出現可以被已觀測到的資料所解釋。
範例：在一個客戶資料函式庫中，某個年齡段的人更傾向於不填寫收入資訊。這種情況下，收入資訊的缺失與年齡相關。
非隨機缺失（MNAR）：資料的缺失與未觀測到的變數有關。這種情況下，缺失值的出現與資料本身的值有關，處理起來較為複雜。
範例：在房屋價格資料集中，如果房屋的建成年份資訊缺失是因為該資訊在資料收集過程中未被納入考量，那麼這種缺失就是非隨機的。

處理缺失值的方法

對於不同型別的缺失值，可以採用以下方法進行處理：

均值、中位數或眾數填補：適用於數值型變數，使用該變數的均值或中位數填補缺失值；對於類別型變數，使用眾數填補。

from pyspark.sql.functions import mean, col

# 計算某列的平均值
mean_value = df.agg(mean(col("column_name"))).collect()[0][0]

# 使用平均值填補缺失值
df = df.fillna(mean_value, subset=["column_name"])

根據模型的填補：使用迴歸模型或其他機器學習模型預測缺失值。
多重填補：透過建立多個不同的填補模型，對每個缺失值產生多個可能的填補值，從而更好地反映不確定性。
使用業務邏輯填補：根據領域知識或業務規則填補缺失值。
範例：假設某房屋價格資料集中，價格資訊的缺失是由於房屋正在拍賣，那麼可以根據業務邏輯將這些房屋的價格統一設為某個預設值（如 $1）。
刪除具有大量缺失值的特徵或樣本：如果某個特徵或樣本的缺失值比例過高，可以考慮刪除該特徵或樣本。

變數選擇

變數選擇是另一個重要的步驟，主要根據以下兩個指標進行：

基數（Cardinality）：指某個變數中不同值的數量。如果某個變數的基數為 1，表示該變數對所有樣本都具有相同的值，因此可以刪除該變數。
```
# 計算每個變數的基數
from pyspark.sql.functions import approxCountDistinct

cardinality = df.select(*[approxCountDistinct(c).alias(c) for c in df.columns])
```

缺失值比例：如果某個變數的缺失值比例過高，可以考慮刪除該變數。

# 計算每個變數的缺失值比例
from pyspark.sql.functions import count, when, isnan, col

missing = df.select(*[count(when(isnan(c) | col(c).isNull(), c)).alias(c) for c in df.columns])

練習題

練習 4-1：缺失值與基數計算

下載墨爾本房屋資料集：https://www.kaggle.com/dansbecker/melbourne-housing-snapshot/data#

計算每個變數的基數，並根據基數篩選變數。

def cardinality_calculation(df, cut_off=1):
    cardinality = df.select(*[approxCountDistinct(c).alias(c) for c in df.columns])
    final_cardinality_df = cardinality.toPandas().transpose()
    final_cardinality_df.reset_index(inplace=True)
    final_cardinality_df.rename(columns={0:'Cardinality'}, inplace=True)
    vars_selected = final_cardinality_df['index'][final_cardinality_df['Cardinality'] <= cut_off]
    return final_cardinality_df, vars_selected

cardinality_df, cardinality_vars_selected = cardinality_calculation(df)

計算每個變數的缺失值比例，並根據缺失值比例篩選變數。

def missing_calculation(df, miss_percentage=0.80):
    missing = df.select(*[count(when(isnan(c) | col(c).isNull(), c)).alias(c) for c in df.columns])
    length_df = df.count()
    final_missing_df = missing.toPandas().transpose()
    final_missing_df.reset_index(inplace=True)
    final_missing_df['missing_percentage'] = final_missing_df[0] / length_df
    vars_to_drop = final_missing_df['index'][final_missing_df['missing_percentage'] > miss_percentage]
    return final_missing_df, vars_to_drop

missing_df, vars_to_drop = missing_calculation(df)

使用中位數填補 YearBuilt 列的缺失值。

from pyspark.sql.functions import median

median_value = df.agg(median(col("YearBuilt"))).collect()[0][0]
df = df.fillna(median_value, subset=["YearBuilt"])

對 BuildingArea 列進行均值填補，並計算填補後的均值。

mean_value = df.agg(mean(col("BuildingArea"))).collect()[0][0]
df = df.fillna(mean_value, subset=["BuildingArea"])
new_mean = df.agg(mean(col("BuildingArea"))).collect()[0][0]
print("New mean after imputation:", new_mean)

載入銀行資料集並進行 EDA

filename = "bank-full.csv"
target_variable_name = "y"

from pyspark.sql import SparkSession
spark = SparkSession.builder.getOrCreate()
df = spark.read.csv(filename, header=True, inferSchema=True, sep=';')
df.show()

# 資料探索
df.count()  # 資料長度
df.describe().toPandas()  # 描述性統計
df.dtypes  # 變數型別
df.printSchema()  # 資料結構
df.groupBy('education').count().show()  # 按教育程度分組統計
df.groupBy(target_variable_name).count().show()  # 目標變數分佈
df.groupBy(['education', target_variable_name]).count().show()  # 多變數分組統計

# 資料聚合
from pyspark.sql.functions import *
df.groupBy(target_variable_name).agg({'balance':'avg', 'age':'avg'}).show()

圖表翻譯：

此圖表呈現了銀行客戶資料集的統計摘要，包括目標變數的分佈、客戶餘額和年齡的平均值等關鍵指標。透過這些資訊，可以初步瞭解客戶群體的基本特徵和潛在趨勢。

資料前處理與特徵工程：類別變數的轉換

在進行資料分析與機器學習模型建置之前，資料前處理是一個非常重要的步驟。其中，類別變數的轉換是常見的需求。本篇文章將介紹如何使用 PySpark 進行類別變數的轉換。

為什麼需要轉換類別變數？

大多數的機器學習演算法都需要數值型的輸入資料。然而，在實際的資料集中，我們常常會遇到類別變數，例如性別、職業、教育程度等。這些類別變數不能直接被機器學習演算法所使用，因此需要進行轉換。

類別變數轉換的方法

PySpark 提供了多種方法來轉換類別變數，包括：

個別欄位轉換：可以個別對每個類別變數進行轉換，但這種方法對於大型資料集來說非常繁瑣。
OneHotEncoder：可以將類別變數轉換為虛擬變數（dummy variables），從而增加特徵的數量。
StringIndexer：可以將類別變數轉換為數值型變數，但需要注意的是，這種方法可能會導致演算法誤解變數之間的關係。
加權索引（WOE）：可以使用目標變數的分佈來建立加權索引，這種方法可以克服 StringIndexer 的缺點。

使用 StringIndexer 進行類別變數轉換

在這裡，我們將使用 StringIndexer 來進行類別變數的轉換。以下是具體步驟：

步驟 1：識別變數型別

def variable_type(df):
    vars_list = df.dtypes
    char_vars = [] 
    num_vars = [] 
    for i in vars_list:
        if i[1] in ('string'):
            char_vars.append(i[0])
        else:
            num_vars.append(i[0])
    return char_vars, num_vars

char_vars, num_vars = variable_type(df)

步驟 2：對類別變數套用 StringIndexer

from pyspark.ml.feature import StringIndexer
from pyspark.ml import Pipeline

def category_to_index(df, char_vars):
    char_df = df.select(char_vars)
    indexers = [StringIndexer(inputCol=c, outputCol=c+"_index", handleInvalid="keep") for c in char_df.columns]
    pipeline = Pipeline(stages=indexers)
    char_labels = pipeline.fit(char_df)
    df = char_labels.transform(df)
    return df, char_labels

df, char_labels = category_to_index(df, char_vars)
df = df.select([c for c in df.columns if c not in char_vars])

#### 內容解密：

variable_type 函式用於區分資料集中的類別變數和數值型變數。
category_to_index 函式使用 StringIndexer 將類別變數轉換為數值型變數。
handleInvalid="keep" 引數確保在遇到新資料時，模型仍能正常運作。
使用 Pipeline 可以批次處理多個類別變數的轉換。

步驟 3：對目標變數套用 StringIndexer

indexer = StringIndexer(inputCol='y', outputCol='y_index', handleInvalid='skip')
indexerModel = indexer.fit(df)
df = indexerModel.transform(df)
df = df.drop(df.y)
df = df.withColumnRenamed('y_index', 'y')

#### 內容解密：

對目標變數 y 使用 StringIndexer 進行轉換。
將轉換後的 y_index 欄位重新命名為 y，並刪除原始的 y 欄位。

步驟 4：組合特徵

from pyspark.ml.feature import VectorAssembler

assembler = VectorAssembler(inputCols=df.columns[:-1], outputCol='features')
df = assembler.transform(df)

#### 內容解密：

使用 VectorAssembler 將所有特徵欄位組合成一個向量欄位 features。
這樣可以方便地將資料輸入到機器學習演算法中。

資料前處理與特徵工程：VectorAssembler 與資料壓縮技術

在機器學習的流程中，資料前處理是非常重要的一環。透過適當的特徵工程，可以提升模型的表現並減少運算成本。本章節將介紹如何使用 PySpark 的 VectorAssembler 來組合多個特徵欄位，以及如何利用主成分分析（PCA）和奇異值分解（SVD）進行資料壓縮。

使用 VectorAssembler 組合特徵

在進行機器學習建模之前，通常需要將多個特徵欄位組合成一個向量，以便進行後續的運算。PySpark 提供了一個方便的工具 VectorAssembler 來實作這一點。

from pyspark.ml.feature import VectorAssembler
from pyspark.ml import Pipeline

def assemble_vectors(df, features_list, target_variable_name):
    stages = []
    # 將多個特徵欄位組合成 'features' 欄位
    assembler = VectorAssembler(inputCols=features_list, outputCol='features')
    stages = [assembler]
    # 選取所有原始欄位、目標變數和新產生的 'features' 欄位
    selectedCols = [target_variable_name, 'features'] + features_list
    pipeline = Pipeline(stages=stages)
    assembleModel = pipeline.fit(df)
    df = assembleModel.transform(df).select(selectedCols)
    return df

# 排除目標變數並選取其他特徵欄位
features_list = df.columns
features_list.remove(target_variable_name)
# 將函式應用於 DataFrame
df = assemble_vectors(df, features_list, target_variable_name)

內容解密：

VectorAssembler 的作用：將多個特徵欄位合併成一個向量欄位 'features'，方便後續的機器學習模型使用。
Pipeline 的使用：將資料處理流程串接起來，確保資料處理的順序性和完整性。
selectedCols 的選取：保留原始特徵欄位、'features' 欄位和目標變數，以便後續分析和建模。

檢視組合後的特徵向量

可以使用以下程式碼檢視 'features' 欄位的結構和內容：

df.schema["features"].metadata["ml_attr"]["attrs"]

並將其轉換為 Pandas DataFrame 以便更直觀的檢視：

import pandas as pd

features_df = pd.DataFrame()
for k, v in df.schema["features"].metadata["ml_attr"]["attrs"].items():
    feature_df_final = pd.DataFrame(v)
    features_df = features_df.append(feature_df_final, ignore_index=True)

內容解密：

檢視 'features' 欄位的 metadata：瞭解 'features' 欄位的結構和組成。
轉換為 Pandas DataFrame：將 Spark DataFrame 的 metadata 轉換為 Pandas DataFrame，以便更方便的檢視和分析。

資料壓縮技術：PCA 和 SVD

在某些情況下，我們需要減少資料的維度，以降低運算成本和提升模型表現。PCA 和 SVD 是兩種常見的資料壓縮技術。

主成分分析（PCA）

PCA 可以用來識別資料中的主要模式，並將資料投影到低維空間中。

@startuml
skinparam backgroundColor #FEFEFE
skinparam componentStyle rectangle

title 資料前處理與特徵工程技術探討

package "資料視覺化流程" {
    package "資料準備" {
        component [資料載入] as load
        component [資料清洗] as clean
        component [資料轉換] as transform
    }

    package "圖表類型" {
        component [折線圖 Line] as line
        component [長條圖 Bar] as bar
        component [散佈圖 Scatter] as scatter
        component [熱力圖 Heatmap] as heatmap
    }

    package "美化輸出" {
        component [樣式設定] as style
        component [標籤註解] as label
        component [匯出儲存] as export
    }
}

load --> clean --> transform
transform --> line
transform --> bar
transform --> scatter
transform --> heatmap
line --> style --> export
bar --> label --> export

note right of scatter
  探索變數關係
  發現異常值
end note

@enduml

圖表翻譯： 此圖示展示了 PCA 的主要步驟，包括資料中心化、計算協方差矩陣、計算特徵值和特徵向量、選取主成分以及資料投影。

PCA 的運作原理

假設我們有一組客戶對汽車滿意度的資料，包括 Quality 和 Reliability 兩個變數。

Participant_ID	Quality	Reliability
1	10	6
2	9	4
3	8	5
4	3	3
5	2	2
6	1	1

透過 PCA，我們可以找出最重要的變數，並將資料投影到一維或二維空間中。

內容解密：

資料中心化：將資料的均值調整為零，以方便後續的運算。
計算協方差矩陣：瞭解不同變數之間的相關性。
計算特徵值和特徵向量：找出資料中的主要模式。
選取主成分：根據特徵值的大小，選取最重要的主成分。
資料投影：將原始資料投影到選取的主成分上，以實作資料壓縮。