超详细解说 sklearn 中的联合特征（FeatureUnion）
1. FeatureUnion 简介
FeatureUnion 是 scikit-learn 中的一个工具，用于并行地组合多个特征提取器的输出。它允许你将不同的特征提取方法（如文本向量化、数值特征缩放、自定义特征工程等）的结果**横向拼接（concatenate）**成一个更大的特征矩阵。
核心思想：
- 并行处理：每个特征提取器独立处理原始数据。
- 特征拼接：将所有提取器的输出在特征维度（列方向）上拼接。
- 统一接口：对外提供与单个转换器相同的
fit、transform、fit_transform接口。
适用场景：
- 同时使用多种特征提取方法（如 TF-IDF + 词袋模型 + 自定义统计特征）。
- 处理异构数据（如文本 + 数值 + 分类特征）。
- 构建复杂特征工程流水线。
2. FeatureUnion 基本语法
from sklearn.pipeline import FeatureUnion
from sklearn.feature_extraction.text import TfidfVectorizer, CountVectorizer
from sklearn.preprocessing import StandardScaler
from sklearn.decomposition import PCA
# 创建多个特征提取器
tfidf = TfidfVectorizer()
bow = CountVectorizer()
scaler = StandardScaler()
# 使用 FeatureUnion 组合
combined_features = FeatureUnion([
('tfidf', tfidf),
('bow', bow),
# ('scaler', scaler) # 注意：scaler 通常用于数值特征，不能直接用于文本
])
# 使用方法
X_transformed = combined_features.fit_transform(X)
!注意：所有特征提取器必须能处理相同的输入数据格式（如都是文本或都是数值数组）。
3.完整代码示例
示例1:文本特征组合(TF-IDF+词袋模型)
from sklearn.pipeline import FeatureUnion
from sklearn.feature_extraction.text import TfidfVectorizer, CountVectorizer
from sklearn.datasets import fetch_20newsgroups
from sklearn.model_selection import train_test_split
from sklearn.linear_model import LogisticRegression
from sklearn.metrics import classification_report
# 加载数据
categories = ['alt.atheism', 'soc.religion.christian']
newsgroups = fetch_20newsgroups(subset='all', categories=categories, remove=('headers', 'footers', 'quotes'))
X, y = newsgroups.data, newsgroups.target
# 划分训练测试集
X_train, X_test, y_train, y_test = train_test_split(X, y, test_size=0.2, random_state=42)
# 创建 FeatureUnion
feature_union = FeatureUnion([
('tfidf', TfidfVectorizer(max_features=1000, stop_words='english')),
('bow', CountVectorizer(max_features=500, ngram_range=(1, 2), stop_words='english'))
])
# 训练特征提取器
X_train_features = feature_union.fit_transform(X_train)
X_test_features = feature_union.transform(X_test)
print(f"训练集特征维度: {X_train_features.shape}") # (样本数, 1500) = 1000(TF-IDF) + 500(BOW)
print(f"测试集特征维度: {X_test_features.shape}")
# 训练分类器
clf = LogisticRegression(random_state=42)
clf.fit(X_train_features, y_train)
# 预测
y_pred = clf.predict(X_test_features)
print("\n分类报告:")
print(classification_report(y_test, y_pred, target_names=newsgroups.target_names))
示例2:数值特征组合(PCA+原始特征)
from sklearn.pipeline import FeatureUnion
from sklearn.decomposition import PCA
from sklearn.preprocessing import StandardScaler
from sklearn.datasets import load_iris
from sklearn.model_selection import train_test_split
from sklearn.ensemble import RandomForestClassifier
import numpy as np
# 加载数据
iris = load_iris()
X, y = iris.data, iris.target
# 划分数据集
X_train, X_test, y_train, y_test = train_test_split(X, y, test_size=0.3, random_state=42)
# 创建 FeatureUnion
feature_union = FeatureUnion([
('pca', PCA(n_components=2)), # 降维到2个主成分
('scaler', StandardScaler()) # 标准化原始特征
])
# 训练并转换
X_train_combined = feature_union.fit_transform(X_train)
X_test_combined = feature_union.transform(X_test)
print(f"原始特征维度: {X_train.shape[1]}") # 4
print(f"组合后特征维度: {X_train_combined.shape[1]}") # 6 = 2(PCA) + 4(标准化)
# 训练分类器
clf = RandomForestClassifier(n_estimators=100, random_state=42)
clf.fit(X_train_combined, y_train)
# 评估
accuracy = clf.score(X_test_combined, y_test)
print(f"\n测试集准确率: {accuracy:.4f}")
示例3:自定义特征提取器+内置提取器
from sklearn.base import BaseEstimator, TransformerMixin
from sklearn.pipeline import FeatureUnion
from sklearn.feature_extraction.text import TfidfVectorizer
import numpy as np
import re
class TextStatsExtractor(BaseEstimator, TransformerMixin):
"""自定义文本统计特征提取器"""
def fit(self, X, y=None):
return self
def transform(self, X):
# 提取文本长度、单词数、大写字母比例等统计特征
features = []
for text in X:
length = len(text)
word_count = len(text.split())
uppercase_ratio = sum(1 for c in text if c.isupper()) / (len(text) + 1e-8)
exclamation_count = text.count('!')
question_count = text.count('?')
features.append([
length,
word_count,
uppercase_ratio,
exclamation_count,
question_count
])
return np.array(features)
# 示例数据
texts = [
"This is a GREAT product!!!",
"I hate this item... it's terrible.",
"Average quality, nothing special.",
"AMAZING!!! Best purchase ever!!!",
"Not bad, but could be better?"
]
# 创建 FeatureUnion
feature_union = FeatureUnion([
('tfidf', TfidfVectorizer(max_features=50, stop_words='english')),
('stats', TextStatsExtractor())
])
# 转换数据
X_combined = feature_union.fit_transform(texts)
print("特征名称:")
feature_names = []
# TF-IDF 特征名
tfidf_names = feature_union.transformer_list[0][1].get_feature_names_out()
feature_names.extend([f"tfidf_{name}" for name in tfidf_names])
# 统计特征名
stat_names = ['length', 'word_count', 'uppercase_ratio', 'exclamation_count', 'question_count']
feature_names.extend(stat_names)
print(f"总特征数: {X_combined.shape[1]}")
print("前5个特征名:", feature_names[:5])
print("后5个特征名:", feature_names[-5:])
print("\n转换后的特征矩阵:")
print(X_combined.toarray())
4.高级用法
4.1设置权重(transformer_weights)
from sklearn.pipeline import FeatureUnion
from sklearn.feature_extraction.text import TfidfVectorizer, CountVectorizer
# 为不同特征提取器设置权重
feature_union = FeatureUnion([
('tfidf', TfidfVectorizer(max_features=100)),
('bow', CountVectorizer(max_features=100))
], transformer_weights={
'tfidf': 1.0,
'bow': 0.5 # 词袋模型的特征乘以0.5
})
# 权重会在 transform 后应用
X_weighted = feature_union.fit_transform(texts)
4.2与Pipeline结合使用
from sklearn.pipeline import Pipeline, FeatureUnion
from sklearn.feature_extraction.text import TfidfVectorizer
from sklearn.preprocessing import StandardScaler
from sklearn.svm import SVC
from sklearn.datasets import fetch_20newsgroups
# 创建复杂的流水线
pipeline = Pipeline([
('features', FeatureUnion([
('tfidf', TfidfVectorizer(max_features=1000)),
('stats', TextStatsExtractor()) # 假设已定义
])),
('scaler', StandardScaler(with_mean=False)), # 稀疏矩阵不能用 with_mean=True
('classifier', SVC(kernel='linear'))
])
# 使用流水线
newsgroups = fetch_20newsgroups(subset='train', categories=['alt.atheism', 'soc.religion.christian'])
pipeline.fit(newsgroups.data, newsgroups.target)
4.3使用make_union快捷方式
from sklearn.pipeline import make_union
from sklearn.decomposition import PCA
from sklearn.feature_selection import SelectKBest
# 简化语法
union = make_union(
PCA(n_components=2),
SelectKBest(k=3),
n_jobs=1 # 并行处理的作业数
)
5.重要注意事项
5.1输入数据一致性
# × 错误：混合不同类型的数据处理
feature_union = FeatureUnion([
('tfidf', TfidfVectorizer()), #处理文本
('scaler', StandardScaler()) #处理数值-会报错！
])
5.2稀疏矩阵处理
from sklearn.feature_extraction.text import TfidfVectorizer
from sklearn.preprocessing import StandardScaler
# TF-IDF 产生稀疏矩阵
tfidf = TfidfVectorizer()
X_sparse = tfidf.fit_transform(texts)
# StandardScaler 不能直接处理稀疏矩阵
# scalar = StandardScaler() # 会报错！
# 解决方案：使用 with_mean=False
scaler=StandardScaler(with_mean=False) # 可以处理稀疏矩阵
5.3 并行处理 (n_jobs)
feature_union = FeatureUnion([
('tfidf1', TfidfVectorizer(max_features=1000)),
('tfidf2', TfidfVectorizer(max_features=1000, ngram_range=(2,2)))
], n_jobs=-1) # 使用所有CPU核心并行处理
6.调试与检查
6.1查看各组件输出维度
feature_union = FeatureUnion([
('tfidf', TfidfVectorizer(max_features=500)),
('bow', CountVectorizer(max_features=300))
])
X_combined = feature_union.fit_transform(texts)
# 检查每个组件的输出
for name, transformer in feature_union.transformer_list:
X_part = transformer.transform(texts)
print(f"{name}: {X_part.shape}")
print(f"Combined: {X_combined.shape}")
6.2 获取特征名称
def get_feature_names(feature_union):
"""获取 FeatureUnion 的所有特征名称"""
feature_names = []
for name, transformer in feature_union.transformer_list:
if hasattr(transformer, 'get_feature_names_out'):
names = transformer.get_feature_names_out()
elif hasattr(transformer, 'get_feature_names'):
names = transformer.get_feature_names()
else:
# 对于自定义转换器，可能需要手动定义
n_features = transformer.transform([texts[0]]).shape[1]
names = [f"{name}_feature_{i}" for i in range(n_features)]
feature_names.extend([f"{name}__{n}" for n in names])
return feature_names
# 使用示例
feature_names = get_feature_names(feature_union)
print(f"总特征数: {len(feature_names)}")
print("前10个特征名:", feature_names[:10])
7. 替代方案：ColumnTransformer
对于异构数据（不同列不同类型），推荐使用 ColumnTransformer：
from sklearn.compose import ColumnTransformer
from sklearn.preprocessing import StandardScaler, OneHotEncoder
from sklearn.feature_extraction.text import TfidfVectorizer
import pandas as pd
# 创建示例数据框
df = pd.DataFrame({
'text': ['good product', 'bad service', 'excellent quality'],
'price': [100, 200, 150],
'category': ['A', 'B', 'A']
})
# 使用 ColumnTransformer 处理不同列
preprocessor = ColumnTransformer([
('text', TfidfVectorizer(), 'text'),
('num', StandardScaler(), ['price']),
('cat', OneHotEncoder(), ['category'])
])
X_transformed = preprocessor.fit_transform(df)
print(f"转换后形状: {X_transformed.shape}")
8. 总结
FeatureUnion 的优势：
- ✅ 简化多特征提取器的组合
- ✅ 提供统一的接口
- ✅ 支持并行处理
- ✅ 可与 Pipeline 无缝集成
使用建议：
- 当需要组合同类型数据的多种特征提取方法时使用
- 对于异构数据，优先考虑 ColumnTransformer
- 注意稀疏矩阵的处理
- 合理设置特征权重
- 使用 make_union 简化代码
最佳实践：
# 推荐的完整流程
from sklearn.pipeline import Pipeline, FeatureUnion
from sklearn.model_selection import cross_val_score
# 1. 定义特征组合
features = FeatureUnion([
('tfidf', TfidfVectorizer(max_features=1000)),
('custom', CustomFeatureExtractor())
])
# 2. 创建完整流水线
pipeline = Pipeline([
('features', features),
('classifier', LogisticRegression())
])
# 3. 交叉验证评估
scores = cross_val_score(pipeline, X, y, cv=5)
print(f"平均准确率: {scores.mean():.4f} (+/- {scores.std() * 2:.4f})")
通过合理使用 FeatureUnion，你可以构建强大的特征工程系统，显著提升机器学习模型的性能！