machine-learning - 如何对实时数据进行特征工程？

Question

我通过以下步骤制作了一个很好的线性回归模型:

1. 数据集成
2. 数据标准化/缩放(数据预处理和特征工程)
3. 模型构建(使用线性回归和 SGD，并使用交叉验证)
4. 测试

我的问题是，如果我们在生产环境中使用这个模型，那么我们如何进行实时数据的特征工程，因为这个模型是通过特征标准化和缩放构建的，那么如何标准化实时数据并进行缩放以获得良好的预测？我们不需要显式的特征工程来进行交叉验证和测试步骤，因为这可以在构建模型之前的数据预处理步骤中完成。 实时数据特征工程怎么样？

Answer 1

此数据非常适合 Featuretools 。它是一个开源自动化特征工程库，明确处理时间以确保不会引入标签泄漏。对于您的音乐数据，您可以创建两个实体:“users”和“artist_plays”，然后应用featuretools.dfs (深度特征合成)生成特征。将实体视为与关系数据库中的表相同。深度特征合成创建一个单表特征矩阵，准备从多个具有高级统计特征的不同表进行建模。这是short post解释它是如何工作的。

此示例使用纯 Python，但可以修改为与 Spark 或 Dask 一起使用。

# Create entityset
import featuretools as ft
from sklearn.preprocessing import Imputer, StandardScaler
import pandas as pd
import pickle

def load_entityset(user_df, artist_plays_df):
    es = ft.EntitySet("artist plays")
    es.entity_from_dataframe("users", user_df, index="user_id")
    es.entity_from_dataframe("artist_plays", artist_plays_df, index="artist_id")
    es.add_relationship(ft.Relationship(es['users']['user_id'], es['artist_plays']['user_id']))
    return es

user_df = pd.read_csv("training_user.csv")
artist_plays_df = pd.read_csv("training_artist_plays.csv")
es = load_entityset(user_df, artist_plays_df)
feature_matrix, feature_defs = ft.dfs(entityset=es,
                                      target_entity='artist_plays',
                                      ignore_variables={'artist_plays': ['play']})

# encode categoricals
encoded_fm, encoded_fl = ft.encode_features(feature_matrix, feature_defs)

# Impute/scale using SKLearn
imputer = Imputer()
scaler = StandardScaler()

imputed_fm = imputer.fit_transform(encoded_fm)
scaled_fm = scaler.fit_transform(imputed_fm)

# Now, save the encoded feature list, and the imputer/scaler to files to reuse in production
ft.save_features(encoded_fl, 'fl.p')
with open('imputer.p', 'wb') as f:
    pickle.dump(imputer, f)
with open('scaler.p', 'wb') as f:
    pickle.dump(scaler, f)

然后在生产中:

import featuretools as ft
import pickle
import pandas as pd

# load previous data
old_user_df = pd.read_csv("training_user.csv")
old_artist_plays_df = pd.read_csv("training_artist_plays.csv")
es_old = load_entityset(old_user_df, old_artist_plays_df)


# load new data
user_df = pd.read_csv("new_user.csv")
artist_plays_df = pd.read_csv("new_artist_plays.csv")
es_updated = load_entityset(user_df, artist_plays_df)

# merge both data sources
es = es_old.concat(es_updated)

# load back in encoded features
features = ft.load_features('fl.p', es)
fm = ft.calculate_feature_matrix(features,
                                 entityset=es,
                                 instance_ids=es_updated['artist_plays'].get_all_instances())

# impute and scale
with open('imputer.p', 'r') as f:
    imputer = pickle.load(f)
imputed_fm = imputer.transform(fm)
with open('scaler.p', 'r') as f:
    scaler = pickle.load(f)
scaled_fm = scaler.transform(imputed_fm)

我们有一些使用此工作流程的演示，请查看 this example预测杂货购物者将来会购买什么。

我还在实时生产环境中使用此工作流程来预测大型软件项目的交付指标 - 查看此 white paper我们已经发布了详细介绍此方法和实时部署结果的文章。