This site is used to record and share the informations & useful tips during the coding and research.
Overview1. 安装顺序Visual Studio2. 安装ta-lib3.安装Zipline3.0 Overview 换了Windows电脑了，最近需要在Windows10系统上面使用Python的Zipline包，折腾了一上午，才终于搞定。必须记录一下整个安装过程，以后方便自己和他人。 先说明一下，github上面这个zipline包zipline最新版本是2020年的1.4.1版本，这个版本最多支持到Python3.6，所以这个老版本的Zipline应该是已经废弃了。最新的可用的应该是Zipline3.0，地址在这里Zipline 3.0。Zipline3.0最高支持到Python3.11。 1. 安装顺序Visual Studio 第一步是安装C++开发环境，Visual Studio，这一步就不介绍了。 2. 安装ta-lib 第二步是安装ta-lib包，这一步的时候，先去ta-lib-0.4.0-msvc.zip下载一个ta-lib-0.4.0-msvc.zip，然后解压，放到C盘根目录。因为这个是32位的，所以我们要手动给他编制成64位的。在开始菜单打开x64 Native Tools Command Prompt for VS 2022，在VS 2022命令行中进入到这个地方C:\ta-lib\c\make\cdr\win32\msvc，输入nmake。 编译完成后，再打开cmd，命令行中(注意本命令行和VS命令行的区别)运行pip install ta-lib。可能会报错如下: LINK : fatal error LNK1327: 运行 rc.exe 期间出错 error: command 'D:\\Program Files (x86)\\Microsoft Visual Studio 14.0\\VC\\BIN\\x86_amd64\\link.exe' failed with exit code 1327 [end of output] note: This error originates from a subprocess, and is likely not a problem with pip. ERROR: Failed building wheel for ta-lib Failed to build ta-lib ERROR: Could not build wheels for ta-lib, which is required to install pyproject.toml-based projects 这是因为D:\Program Files (x86)\Microsoft Visual Studio 14.0\VC\bin这里缺少rc.exe和rcdll.dll文件。我们从C:\Program Files (x86)\Windows Kits\10\bin\10.0.18362.0\x64文件夹中找到这俩文件，复制到D:\Program Files (x86)\Microsoft Visual Studio 14.0\VC\bin文件夹。 继续执行pip install ta-lib就安装好了。 3.安装Zipline3.0 在cmd中运行以下命令pip install zipline-reloaded，即可安装成功。假如没有前两步，直接安装Zipline3.0，那么会报缺少ta_libc.h。 至此，zipline就安装好了。 本文参考了github，Stack Overflow和以下文章，鸣谢。 重装Anaconda,指定虚拟环境路径，清华源，Jupyter，zipline Failed to build ta-lib ERROR: Could not build wheels for ta-lib, which is required to install pyproject.toml-based project #566 Failed to install ta-lib #127 Windows系统下安装TA_Lib教程
Overview1. 加载并处理数据2. 建模训练2.1 输入层2.2 Cross网络2.3 Deep网络2.4 输出层2.5 模型编译2.6 训练模型 Overview 之前的文章，我们记录了如何用TensorFlow 2.0中的Keras模块实现DeepFM算法，TensorFlow 2.0实战DeepFM。本文继续用TensorFlow 2.0来实现另一个常见的深度学习推荐算法Deep&Cross。 1. 加载并处理数据 依然沿用之前的1,000,000条criteo数据。 import numpy as np import pandas as pd from sklearn.preprocessing import LabelEncoder, MinMaxScaler import tensorflow as tf from tensorflow.keras import regularizers from tensorflow.keras.layers import * from tensorflow.keras.models import Model from tensorflow.keras.utils import plot_model import tensorflow.keras.backend as K from tensorflow.keras.callbacks import TensorBoard import matplotlib.pyplot as plt %matplotlib inline data = pd.read_csv('criteo/criteo_sampled_data.csv', sep='\t', header=None) data.head() label = ['label'] dense_features = ['I' + str(i) for i in range(1, 14)] sparse_features = ['C' + str(i) for i in range(1, 27)] name_list = label + dense_features + sparse_features data.columns = name_list 处理连续型特征： # 数值型特征空值填0 data[dense_features] = data[dense_features].fillna(0) # 数值型特征归一化 scaler = MinMaxScaler(feature_range=(0, 1)) data[dense_features] = scaler.fit_transform(data[dense_features]) 处理稀疏类别特征： data[sparse_features] = data[sparse_features].fillna("-1") for feat in sparse_features: lbe = LabelEncoder() data[feat] = lbe.fit_transform(data[feat]) 2. 建模训练 我们先来看看DCN算法的架构图： 2.1 输入层 # 稠密特征 dense_inputs = [] for fea in dense_features: _input = Input([1], name=fea) dense_inputs.append(_input) dense_inputs 输出13个稠密特征的张量： [<tf.Tensor 'I1:0' shape=(None, 1) dtype=float32>, <tf.Tensor 'I2:0' shape=(None, 1) dtype=float32>, <tf.Tensor 'I3:0' shape=(None, 1) dtype=float32>, <tf.Tensor 'I4:0' shape=(None, 1) dtype=float32>, <tf.Tensor 'I5:0' shape=(None, 1) dtype=float32>, <tf.Tensor 'I6:0' shape=(None, 1) dtype=float32>, <tf.Tensor 'I7:0' shape=(None, 1) dtype=float32>, <tf.Tensor 'I8:0' shape=(None, 1) dtype=float32>, <tf.Tensor 'I9:0' shape=(None, 1) dtype=float32>, <tf.Tensor 'I10:0' shape=(None, 1) dtype=float32>, <tf.Tensor 'I11:0' shape=(None, 1) dtype=float32>, <tf.Tensor 'I12:0' shape=(None, 1) dtype=float32>, <tf.Tensor 'I13:0' shape=(None, 1) dtype=float32>] 稠密特征是不用做Embedding的，我们直接连接起来： concat_dense_inputs = Concatenate(axis=1)(dense_inputs) 然后处理稀疏特征，对其做Embedding,我们假设每个特征映射成为8维的向量： sparse_inputs = [] for fea in sparse_features: _input = Input([1], name=fea) sparse_inputs.append(_input) k = 8 sparse_kd_embed = [] for _input in sparse_inputs: f = _input.name.split(':')[0] voc_size = data[f].nunique() _embed = Flatten()(Embedding(voc_size+1, k, embeddings_regularizer=regularizers.l2(0.7))(_input)) sparse_kd_embed.append(_embed) sparse_kd_embed 可以看到Embedding之后的结果： [<tf.Tensor 'flatten/Identity:0' shape=(None, 8) dtype=float32>, <tf.Tensor 'flatten_1/Identity:0' shape=(None, 8) dtype=float32>, <tf.Tensor 'flatten_2/Identity:0' shape=(None, 8) dtype=float32>, <tf.Tensor 'flatten_3/Identity:0' shape=(None, 8) dtype=float32>, <tf.Tensor 'flatten_4/Identity:0' shape=(None, 8) dtype=float32>, <tf.Tensor 'flatten_5/Identity:0' shape=(None, 8) dtype=float32>, <tf.Tensor 'flatten_6/Identity:0' shape=(None, 8) dtype=float32>, <tf.Tensor 'flatten_7/Identity:0' shape=(None, 8) dtype=float32>, <tf.Tensor 'flatten_8/Identity:0' shape=(None, 8) dtype=float32>, <tf.Tensor 'flatten_9/Identity:0' shape=(None, 8) dtype=float32>, <tf.Tensor 'flatten_10/Identity:0' shape=(None, 8) dtype=float32>, <tf.Tensor 'flatten_11/Identity:0' shape=(None, 8) dtype=float32>, <tf.Tensor 'flatten_12/Identity:0' shape=(None, 8) dtype=float32>, <tf.Tensor 'flatten_13/Identity:0' shape=(None, 8) dtype=float32>, <tf.Tensor 'flatten_14/Identity:0' shape=(None, 8) dtype=float32>, <tf.Tensor 'flatten_15/Identity:0' shape=(None, 8) dtype=float32>, <tf.Tensor 'flatten_16/Identity:0' shape=(None, 8) dtype=float32>, <tf.Tensor 'flatten_17/Identity:0' shape=(None, 8) dtype=float32>, <tf.Tensor 'flatten_18/Identity:0' shape=(None, 8) dtype=float32>, <tf.Tensor 'flatten_19/Identity:0' shape=(None, 8) dtype=float32>, <tf.Tensor 'flatten_20/Identity:0' shape=(None, 8) dtype=float32>, <tf.Tensor 'flatten_21/Identity:0' shape=(None, 8) dtype=float32>, <tf.Tensor 'flatten_22/Identity:0' shape=(None, 8) dtype=float32>, <tf.Tensor 'flatten_23/Identity:0' shape=(None, 8) dtype=float32>, <tf.Tensor 'flatten_24/Identity:0' shape=(None, 8) dtype=float32>, <tf.Tensor 'flatten_25/Identity:0' shape=(None, 8) dtype=float32>] 连接起来稀疏特征映射结果： concat_sparse_inputs = Concatenate(axis=1)(sparse_kd_embed) 最后，把稠密特征和稀疏特征两部分结果合并，成为整个Deep&Cross算法的Embedding层和连接层，这部分是Cross网络和Deep网络两部分共用的，这一点和DeepFM算法有明显的区别。 embed_inputs = Concatenate(axis=1)([concat_sparse_inputs, concat_dense_inputs]) 2.2 Cross网络 先来一个Cross层的构建函数： def cross_layer(x0, xl): """ 实现一层cross layer @param x0: 前面得到的embedding层 @param xl: 第L层的输出结果,注意L和1的区别 """ # 获取xL层的embedding size embed_dim = xl.shape[-1] # 初始化当前层的W和b w = tf.Variable(tf.random.truncated_normal(shape=(embed_dim,), stddev=0.01)) b = tf.Variable(tf.zeros(shape=(embed_dim,))) # 计算feature crossing x_lw = tf.tensordot(tf.reshape(xl, [-1, 1, embed_dim]), w, axes=1) cross = x0 * x_lw return cross + b + xl 这里我们需要知道，在算$x_{0}x_{l}^{T}W_{l}$时一定要先算$x_{l}^{T}W_{l}$，这样可以节约内存。 然后逐层构建Cross层： def build_cross_layer(x0, num_layer=3): """ 构建多层cross layer @param x0: 前面得到的embedding层 @param num_layers: cross net的层数 """ # 初始化xl为x0 xl = x0 # 构建多层cross net for i in range(num_layer): xl = cross_layer(x0, xl) return xl 开始构建Cross层： cross_layer_output = build_cross_layer(embed_inputs, 3) 2.3 Deep网络 这一步，会和Cross部分共用前面得到的embedding结果。 fc_layer = Dropout(0.5)(Dense(128, activation='relu')(embed_inputs)) # Deep层和Cross层共用embedding结果 fc_layer = Dropout(0.3)(Dense(128, activation='relu')(fc_layer)) fc_layer_output = Dropout(0.1)(Dense(128, activation='relu')(fc_layer)) 2.4 输出层 我们将Deep部分和Cross部分结果Stacking一下： stack_layer = Concatenate()([cross_layer_output, fc_layer_output]) output_layer = Dense(1, activation='sigmoid', use_bias=True)(stack_layer) 2.5 模型编译 建模查看模型结构： model = Model(dense_inputs + sparse_inputs, output_layer) plot_model(model, "dcn.png") model.summary() 编译模型： model.compile(optimizer="adam", loss="binary_crossentropy", metrics=["binary_crossentropy", metrics.AUC(name='auc')]) 2.6 训练模型 tbCallBack = TensorBoard(log_dir='./logs', histogram_freq=0, write_graph=True, # write_grads=True, write_images=True, embeddings_freq=0, embeddings_layer_names=None, embeddings_metadata=None) train_data = data.loc[:800000-1] valid_data = data.loc[800000:] train_dense_x = [train_data[f].values for f in dense_features] train_sparse_x = [train_data[f].values for f in sparse_features] train_label = [train_data['label'].values] val_dense_x = [valid_data[f].values for f in dense_features] val_sparse_x = [valid_data[f].values for f in sparse_features] val_label = [valid_data['label'].values] model.fit(train_dense_x+train_sparse_x, train_label, epochs=5, batch_size=128, validation_data=(val_dense_x+val_sparse_x, val_label), callbacks=[tbCallBack] ) 训练结果： Epoch 1/5 6250/6250 [==============================] - 309s 49ms/step - loss: 14.1232 - binary_crossentropy: 0.5130 - auc: 0.7139 - val_loss: 0.6548 - val_binary_crossentropy: 0.5039 - val_auc: 0.7356 Epoch 2/5 6250/6250 [==============================] - 307s 49ms/step - loss: 0.6697 - binary_crossentropy: 0.5056 - auc: 0.7259 - val_loss: 0.6767 - val_binary_crossentropy: 0.4949 - val_auc: 0.7364 Epoch 3/5 6250/6250 [==============================] - 307s 49ms/step - loss: 0.6805 - binary_crossentropy: 0.5036 - auc: 0.7291 - val_loss: 0.6862 - val_binary_crossentropy: 0.4947 - val_auc: 0.7371 Epoch 4/5 6250/6250 [==============================] - 308s 49ms/step - loss: 0.6845 - binary_crossentropy: 0.5025 - auc: 0.7308 - val_loss: 0.6864 - val_binary_crossentropy: 0.4962 - val_auc: 0.7392 Epoch 5/5 6250/6250 [==============================] - 307s 49ms/step - loss: 0.6888 - binary_crossentropy: 0.5018 - auc: 0.7317 - val_loss: 0.6823 - val_binary_crossentropy: 0.4942 - val_auc: 0.7378 至此，Deep&Cross算法已经用TensorFlow 2.0实现。总体来说，这个算法实现起来，比DeepFM要简单一些。 本文主要参考了以下文章： CTR模型代码实战 CTR预估模型：DeepFM/Deep&Cross/xDeepFM/AutoInt代码实战与讲解
Overview1. 准备数据2. 加载并处理数据2.1 处理连续型特征2.2 处理稀疏类别特征3. 建模训练3.1 处理一阶特征3.2 二阶特征3.3 DNN层3.4 合并输出层3.5 编译模型 Overview 推荐系统最核心的部分在于排序算法，也就是我们说的CTR预估问题。这部分算法在最近几年发展迅速，我们这篇文章，希望用经典的criteo数据集，和TensorFlow 2.0中的Keras模块，来演示一下DeepFM是怎么做CTR预估的。 1. 准备数据 在这里下载criteo数据集：Download Kaggle Display Advertising Challenge Dataset。 或者直接在终端下载,然后解压： mkdir criteo cd criteo wget https://s3-eu-west-1.amazonaws.com/kaggle-display-advertising-challenge-dataset/dac.tar.gz tar -zxvf dac.tar.gz 这个数据集，训练集40,000,000行，从readme.txt中可以知道数值型特征的有13个，类别型特征有26个，特征名称未提供，样本按时间排序。测试集6,000,000行，不含label。解压后，可以看到训练集是11.1G，测试集是1.46G。当前这个单机是打不开的。那我们就读取1,000,000条训练集好了。 head -n 1000000 train.txt > criteo_sampled_data.csv cd .. 2. 加载并处理数据 首先把抽样过的数据加载到内存： import numpy as np import pandas as pd from sklearn.preprocessing import LabelEncoder, MinMaxScaler import tensorflow as tf from tensorflow.keras import regularizers from tensorflow.keras.layers import * from tensorflow.keras.models import Model from tensorflow.keras.utils import plot_model import tensorflow.keras.backend as K from tensorflow.keras.callbacks import TensorBoard import matplotlib.pyplot as plt %matplotlib inline data = pd.read_csv('criteo/criteo_sampled_data.csv', sep='\t', header=None) data.head() 我们给数据加上列名： label = ['label'] dense_features = ['I' + str(i) for i in range(1, 14)] sparse_features = ['C' + str(i) for i in range(1, 27)] name_list = label + dense_features + sparse_features data.columns = name_list data.head() 2.1 处理连续型特征 # 数值型特征空值填0 data[dense_features] = data[dense_features].fillna(0) # 数值型特征归一化 scaler = MinMaxScaler(feature_range=(0, 1)) data[dense_features] = scaler.fit_transform(data[dense_features]) 2.2 处理稀疏类别特征 data[sparse_features] = data[sparse_features].fillna("-1") for feat in sparse_features: lbe = LabelEncoder() data[feat] = lbe.fit_transform(data[feat]) 3. 建模训练 这个部分就是DeepFM模型的精髓所在。算法原理如图所示。 DeepFM模型是对Wide&Deep模型的改进。左边FM层和右边DNN层共享了Embedding层。 3.1 处理一阶特征 首先，数值型特征可以直接变成tensorflow.keras.layers.Input单元。 dense_inputs = [] for fea in dense_features: _input = Input([1], name=fea) dense_inputs.append(_input) dense_inputs 数值型特征的生成的输入单元如下： [<tf.Tensor 'I1:0' shape=(None, 1) dtype=float32>, <tf.Tensor 'I2:0' shape=(None, 1) dtype=float32>, <tf.Tensor 'I3:0' shape=(None, 1) dtype=float32>, <tf.Tensor 'I4:0' shape=(None, 1) dtype=float32>, <tf.Tensor 'I5:0' shape=(None, 1) dtype=float32>, <tf.Tensor 'I6:0' shape=(None, 1) dtype=float32>, <tf.Tensor 'I7:0' shape=(None, 1) dtype=float32>, <tf.Tensor 'I8:0' shape=(None, 1) dtype=float32>, <tf.Tensor 'I9:0' shape=(None, 1) dtype=float32>, <tf.Tensor 'I10:0' shape=(None, 1) dtype=float32>, <tf.Tensor 'I11:0' shape=(None, 1) dtype=float32>, <tf.Tensor 'I12:0' shape=(None, 1) dtype=float32>, <tf.Tensor 'I13:0' shape=(None, 1) dtype=float32>] 连接成为稠密层： concat_dense_inputs = Concatenate(axis=1)(dense_inputs) first_order_dense_layer = Dense(1)(concat_dense_inputs) 其次，处理稀疏类别型特征。 sparse_inputs = [] for fea in sparse_features: _input = Input([1], name=fea) sparse_inputs.append(_input) 我们需要将稀疏特征Embedding成为稠密特征，需要用到tensorflow.keras.layers.Embedding模块，然后将其打平，这里是第一次对稀疏特征做Embedding，且映射成为长度为1的向量（还有一次映射）： sparse_1d_embed = [] for i, _input in enumerate(sparse_inputs): f = sparse_features[i] voc_size = data[f].nunique() _embed = Flatten()(Embedding(voc_size+1, 1, embeddings_regularizer=regularizers.l2(0.5))(_input)) sparse_1d_embed.append(_embed) sparse_1d_embed 稀疏特征的生成的输入单元如下： [<tf.Tensor 'flatten/Identity:0' shape=(None, 1) dtype=float32>, <tf.Tensor 'flatten_1/Identity:0' shape=(None, 1) dtype=float32>, <tf.Tensor 'flatten_2/Identity:0' shape=(None, 1) dtype=float32>, <tf.Tensor 'flatten_3/Identity:0' shape=(None, 1) dtype=float32>, <tf.Tensor 'flatten_4/Identity:0' shape=(None, 1) dtype=float32>, <tf.Tensor 'flatten_5/Identity:0' shape=(None, 1) dtype=float32>, <tf.Tensor 'flatten_6/Identity:0' shape=(None, 1) dtype=float32>, <tf.Tensor 'flatten_7/Identity:0' shape=(None, 1) dtype=float32>, <tf.Tensor 'flatten_8/Identity:0' shape=(None, 1) dtype=float32>, <tf.Tensor 'flatten_9/Identity:0' shape=(None, 1) dtype=float32>, <tf.Tensor 'flatten_10/Identity:0' shape=(None, 1) dtype=float32>, <tf.Tensor 'flatten_11/Identity:0' shape=(None, 1) dtype=float32>, <tf.Tensor 'flatten_12/Identity:0' shape=(None, 1) dtype=float32>, <tf.Tensor 'flatten_13/Identity:0' shape=(None, 1) dtype=float32>, <tf.Tensor 'flatten_14/Identity:0' shape=(None, 1) dtype=float32>, <tf.Tensor 'flatten_15/Identity:0' shape=(None, 1) dtype=float32>, <tf.Tensor 'flatten_16/Identity:0' shape=(None, 1) dtype=float32>, <tf.Tensor 'flatten_17/Identity:0' shape=(None, 1) dtype=float32>, <tf.Tensor 'flatten_18/Identity:0' shape=(None, 1) dtype=float32>, <tf.Tensor 'flatten_19/Identity:0' shape=(None, 1) dtype=float32>, <tf.Tensor 'flatten_20/Identity:0' shape=(None, 1) dtype=float32>, <tf.Tensor 'flatten_21/Identity:0' shape=(None, 1) dtype=float32>, <tf.Tensor 'flatten_22/Identity:0' shape=(None, 1) dtype=float32>, <tf.Tensor 'flatten_23/Identity:0' shape=(None, 1) dtype=float32>, <tf.Tensor 'flatten_24/Identity:0' shape=(None, 1) dtype=float32>, <tf.Tensor 'flatten_25/Identity:0' shape=(None, 1) dtype=float32>] 累加起来： first_order_sparse_layer = Add()(sparse_1d_embed) 线性部分合并，即稠密层单元和稀疏层一维映射过的单元继续累加起来： linear_part = Add()([first_order_dense_layer, first_order_sparse_layer]) 3.2 二阶特征 这里，我们只考虑稀疏特征映射后的二阶交叉，第二次对稀疏特征进行Embedding，映射到8维。其二阶FM的公式为： $$ \sum_{i=1}^n\sum_{j=i+1}^n \langle v_i, v_j \rangle x_i x_j=\frac{1}{2} \sum_{f=1}^k[(\sum_{i=1}^n V_{if})^2 - \sum_{i=1}^n V_{if}^2] $$ sparse_kd_embed = [] for i, _input in enumerate(sparse_inputs): f = sparse_features[i] voc_size = data[f].nunique() _embed = Embedding(voc_size+1, k, embeddings_regularizer=regularizers.l2(0.7))(_input) # 这里不要Flatten打平 sparse_kd_embed.append(_embed) 第一步，把所有的8维向量拼接起来： concat_sparse_kd_embed = Concatenate(axis=1)(sparse_kd_embed) 得到一个（26，8）的矩阵。 第二步，对该矩阵进行求和的平方： sum_kd_embed = Lambda(lambda x: K.sum(x, axis=1))(concat_sparse_kd_embed) square_sum_kd_embed = Multiply()([sum_kd_embed, sum_kd_embed]) 第三步，对该矩阵进行求平方和： square_kd_embed = Multiply()([concat_sparse_kd_embed, concat_sparse_kd_embed]) sum_square_kd_embed = Lambda(lambda x: K.sum(x, axis=1))(square_kd_embed) 第四步，求差除以2，再求和。 sub = Subtract()([square_sum_kd_embed, sum_square_kd_embed]) sub = Lambda(lambda x: x*0.5)(sub) second_order_sparse_layer = Lambda(lambda x: K.sum(x, axis=1, keepdims=True))(sub) 至此，我们的FM部分完成。 3.3 DNN层 首先，我们把第二次Embedding之后的26个8维向量打平。 flatten_sparse_embed = Flatten()(concat_sparse_kd_embed) 然后，添加hidden layer并加上dropout： fc_layer = Dropout(0.5)(Dense(256, activation='relu')(flatten_sparse_embed)) fc_layer = Dropout(0.3)(Dense(256, activation='relu')(fc_layer)) fc_layer = Dropout(0.1)(Dense(256, activation='relu')(fc_layer)) fc_layer_output = Dense(1)(fc_layer) 至此，DNN层结束。 3.4 合并输出层 我们把一阶部分和二阶部分，DNN输出层这三部分合并累加，然后用sigmoid函数将其和变成一个概率值。 output_layer = Add()([linear_part, second_order_sparse_layer, fc_layer_output]) output_layer = Activation("sigmoid")(output_layer) 3.5 编译模型 模型构建： model = Model(dense_inputs+sparse_inputs, output_layer) 打印出各单元和节点的关系图： plot_model(model, "deepfm.png") 查看网络结构： model.summary() 编译模型： model.compile(optimizer="adam", loss="binary_crossentropy", metrics=["binary_crossentropy", tf.keras.metrics.AUC(name='auc')]) tbCallBack = TensorBoard(log_dir='./logs', # log 目录 histogram_freq=0, write_graph=True, # write_grads=True, write_images=True, embeddings_freq=0, embeddings_layer_names=None, embeddings_metadata=None) 训练集和验证集按照8:2的比例划分： train_data = data.loc[:800000-1] valid_data = data.loc[800000:] 处理稠密特征和稀疏特征，一定要分开处理： train_dense_x = [train_data[f].values for f in dense_features] train_sparse_x = [train_data[f].values for f in sparse_features] train_label = [train_data['label'].values] val_dense_x = [valid_data[f].values for f in dense_features] val_sparse_x = [valid_data[f].values for f in sparse_features] val_label = [valid_data['label'].values] 模型训练： model.fit(train_dense_x+train_sparse_x, train_label, epochs=5, batch_size=256, validation_data=(val_dense_x+val_sparse_x, val_label), callbacks=[tbCallBack] ) 输出训练过程： Epoch 1/5 3125/3125 [==============================] - 182s 58ms/step - loss: 30.0034 - binary_crossentropy: 0.5280 - auc: 0.6819 - val_loss: 0.6106 - val_binary_crossentropy: 0.5108 - val_auc: 0.7098 Epoch 2/5 3125/3125 [==============================] - 181s 58ms/step - loss: 0.6243 - binary_crossentropy: 0.5168 - auc: 0.7051 - val_loss: 0.6414 - val_binary_crossentropy: 0.5092 - val_auc: 0.7157 Epoch 3/5 3125/3125 [==============================] - 181s 58ms/step - loss: 0.6395 - binary_crossentropy: 0.5135 - auc: 0.7116 - val_loss: 0.6570 - val_binary_crossentropy: 0.5126 - val_auc: 0.7205 Epoch 4/5 3125/3125 [==============================] - 180s 58ms/step - loss: 0.6475 - binary_crossentropy: 0.5123 - auc: 0.7143 - val_loss: 0.6674 - val_binary_crossentropy: 0.5059 - val_auc: 0.7229 Epoch 5/5 3125/3125 [==============================] - 181s 58ms/step - loss: 0.6463 - binary_crossentropy: 0.5113 - auc: 0.7165 - val_loss: 0.6479 - val_binary_crossentropy: 0.5055 - val_auc: 0.7247 至此，DeepFM已经用TensorFlow 2.0实现。 本文主要参考了以下文章： CTR预估模型：DeepFM/Deep&Cross/xDeepFM/AutoInt代码实战与讲解 CTR模型代码实战
Overview1. XGBoost2. LightGBM3. TensorFlow2.0 Overview 之前的文章介绍了一些推荐系统的理论知识，理论虽然很容易理解，但是实际项目当中会有一些细节问题是需要我们去解决的。比如在很多流量很大的电商以及资讯网站的推荐系统中，每天的数据其实是增长很快的，所以模型迭代的频率也是非常高的，甚至于需要每天更新。那么我们就要面临一个很现实的问题，我们每天更新模型，那么每次都要用全量数据去训练模型吗？显然这是不可能的。一来耗费资源，二来时间也不允许。那么，增量学习Incremental Learning就派上用场了。增量学习是这几年才出来的概念，其目的在于训练新数据的同时也能保留以前模型的效果，即在之前的模型基础上对新增的数据进行训练。 主流的几种机器学习框架，已经实现了增量学习的功能，接下来看一下他们是怎么做到的。 1. XGBoost 我们一般情况下用XGBoost训练一个全新的模型，其相关方法是这样的： watchlist = [(xgb_train, 'train'), (xgb_val, 'val'), (xgb_test, 'test')] # training model_path = 'init.model' model = xgb.train(params, xgb_train, num_boost, watchlist) model.save_model(model_path) 如果我们线上已经有一个效果很好的模型，需要在此基础上对新增的数据进行增量训练，那么XGBoost是这样实现的： # 指定基础模型参数，xgb_model # 传入新的增量训练数据，xgb_train_new model_new = xgb.train(params, xgb_train_new, num_boost, xgb_model=model_path) 我们来看这个参数的作用： xgb_model : file name of stored xgb model or 'Booster' instance Xgb model to be loaded before training (allows training continuation). 即在指定的模型上继续训练。 更多相关知识请参考： How can I implement incremental training for xgboost? dmlc/xgboost 2. LightGBM model = lgb.train(params, lgb_train, num_boost_round=1000, valid_sets=lgb_eval, feature_name=x_cols, early_stopping_rounds=10, verbose_eval=False, init_model=model, # 如果init_model不为None，那么就是在init_model基础上接着训练 keep_training_booster=True) # 增量训练 LightGBM实现增量训练也很简单，只需要设置init_model为当前线上的模型，且keep_training_booster设置为True即可。 更多相关知识请参考： Incremental learning using Dataset.subset LightGBM 2.1.1 python API Standard outputs infos even when disabled lightgbm.train 3. TensorFlow2.0 我们知道，TensorFlow2.0已经集成好了Keras高级API，所以，我们看看增量学习在Keras当中是如何实现的。 # 加载线上的模型 model_path = 'init.model' loaded_model = tf.keras.models.load_model(model_path) # 新数据上接着训练 history = loaded_model.fit( train_data_gen, steps_per_epoch=total_train // batch_size, epochs=epochs, validation_data=val_data_gen, validation_steps=total_val // batch_size ) 可以看到，TensorFlow2.0对于增量训练支持也是很完备的。
Overview1. jieba分词工具2. Word2Vec2.1 Spark Word2Vec2.2 NLP ChineseEmbedding3. Top N相似近邻搜索3.1 scipy余弦相似度3.2 LSH局部敏感哈希 Overview 内容相似推荐算法，是推荐系统当中的一个常用的算法。假设我们做的是文章推荐，那么，我们该怎样求两篇文章的相似度呢？本文将记录几个常用工具的使用方法。 1. jieba分词工具 可以说是当前最好用的中文分词组件。我们拿到文章的标题、摘要、全文之后，可以用jieba提取关键词，然后去另一篇文章当中做Doc2Vec，把另一篇文章转化为一个向量，可以用平均或者加权平均的方式得到一个文章的向量。 首先安装这个组件： pip3 install jieba 之后就可以直接导入使用了： import jieba seg_list = jieba.cut("我正在做推荐系统当中的内容相似推荐部分。", cut_all=False) print("Default Mode: " + "/ ".join(seg_list)) # 精确模式 输出如下： Default Mode: 我/ 正在/ 做/ 推荐/ 系统/ 当中/ 的/ 内容/ 相似/ 推荐/ 部分/ 。 直接分词会把所有的词都得到，这样显然是不符合要求的。我们可以提取关键词功能来获得最重要的K个关键词。 import jieba.analyse key_words = jieba.analyse.extract_tags("我正在做推荐系统当中的内容相似推荐部分。", topK=3, withWeight=True, allowPOS=()) print(key_words) 显示如下： [('推荐', 1.386500847515), ('当中', 0.82161213919375), ('相似', 0.80830253373)] 我们会得到一个带权重的关键词列表。 这种只能对文章进行精确匹配，鲁棒性稍微差点。因此我们更需要下面这种更强大的工具Word2Vec。 2. Word2Vec Word2Vec其实是一个神经网络模型。我们不用自己实现，可以使用开源届的现有工具，Spark Word2Vec或腾讯Word2Vec工具NLP ChineseEmbedding。 2.1 Spark Word2Vec 具体用法如下： from pyspark.sql import SparkSession from pyspark.sql import functions as F from pyspark.ml.feature import Word2Vec spark = SparkSession.builder.appName("Spark Word2Vec") \ .master("spark://***.***.***.***:7077") \ .config("spark.cores.max", "10") \ .config("spark.driver.memory", "2G") \ .config("spark.executor.memory", "2G") \ .config("spark.executor.cores", "2")\ .getOrCreate() # 输入数据可以是一个词袋、一个句子或者一个文档 documentDF = spark.createDataFrame([ ("我 正在 做 推荐 系统 当中 的 内容 相似 推荐 部分".split(" "), ), ("我 喜欢 做 推荐 系统".split(" "), ), ("做 机器 学习 我 是 专业 的".split(" "), ) ], ["text"]) # 学习一个词到向量的映射模型 word2Vec = Word2Vec(vectorSize=3, minCount=0, inputCol="text", outputCol="result") model = word2Vec.fit(documentDF) result = model.transform(documentDF) for row in result.collect(): text, vector = row print("Text: [%s] => \nVector: %s\n" % (", ".join(text), str(vector))) 最后我们得到如下答案： Text: [我, 正在, 做, 推荐, 系统, 当中, 的, 内容, 相似, 推荐, 部分] => Vector: [0.02447097257456996,-0.05411485582590103,0.02040436087091538] Text: [我, 喜欢, 做, 推荐, 系统] => Vector: [0.04216975644230843,-0.0064331904053688055,-0.011528981802985073] Text: [做, 机器, 学习, 我, 是, 专业, 的] => Vector: [0.019645389967731065,0.010864340301070894,-0.0007330668491444417] 更多内容可以参考Spark官方文档Extracting, transforming and selecting features。 这个虽然强大，但是受制于我们手头的语料，即训练数据不够充分，所以我们不建议自己训练一个Word2Vec模型，我们更推荐用更强大的腾讯Word2Vec——NLP ChineseEmbedding。 2.2 NLP ChineseEmbedding 这个工具非常有名，具体介绍可以看官网Tencent AI Lab Embedding Corpus for Chinese Words and Phrases。我们这里只记录它是怎么用的。 首先，我们下载到本地，然后解压缩： wget https://ai.tencent.com/ailab/nlp/zh/data/Tencent_AILab_ChineseEmbedding .tar.gz tar -zxvf Tencent_AILab_ChineseEmbedding.tar.gz 得到Tencent_AILab_ChineseEmbedding.txt这个文件。 然后我们需要安装gensim包来辅助我们使用腾讯的word2vec模型。 pip3 install gensim from gensim.models import KeyedVectors wv_from_text = KeyedVectors.load_word2vec_format('Tencent_AILab_ChineseEmbedding.txt', binary=False) model=wv_from_text.wv 这样，我们就得到了这个Word2Vec模型。 可以通过model.word_vec方法将文字转成向量。 3. Top N相似近邻搜索 在我们将文字转成向量之后，可以比较两个向量的相似度，进而选取最相似的N个向量。而相似近邻搜索主要有两种方式：scipy余弦相似度和LSH局部敏感哈希。 3.1 scipy余弦相似度 scipy.spatial.distance.cosine这个方法在scipy这个包里，这个包对于做机器学习的人来说已经很熟悉了，不再赘述。 from scipy.spatial import distance distance.cosine([1, 0, 0], [0, 1, 0]) 这个方式会有问题，每两个物品都对比余弦相似度，那么时间复杂度为$O(N^{2})$，这在推荐系统中是不允许的. 3.2 LSH局部敏感哈希 Locality Sensitive Hashing我们可以从Spark文档当中找到介绍 Locality Sensitive Hashing。 原理是将备选物品哈希到M个分桶当中，然后计算每个分桶里面物品的相似物品。这样性能就提升了很多。 这样，从算法原理上就完成了内容相似推荐。
Overview1. 加载IMDB评论数据2. 构建RNN模型3. 模型编译训练 Overview 本篇文章我们记录一下怎么用TensorFlow 2.0当中的Keras模块来进行RNN和LSTM文本分类。 1. 加载IMDB评论数据 import tensorflow_datasets as tfds import tensorflow as tf from tensorflow.keras import models, layers, losses, optimizers, Sequential import matplotlib.pyplot as plt dataset, info = tfds.load('imdb_reviews/subwords8k', with_info=True, as_supervised=True) train_dataset, test_dataset = dataset['train'], dataset['test'] encoder = info.features['text'].encoder print('Vocabulary size: {}'.format(encoder.vocab_size)) BUFFER_SIZE = 10000 BATCH_SIZE = 64 train_dataset = train_dataset.shuffle(BUFFER_SIZE) train_dataset = train_dataset.padded_batch(BATCH_SIZE) test_dataset = test_dataset.padded_batch(BATCH_SIZE) 2. 构建RNN模型 model = Sequential([ layers.Embedding(encoder.vocab_size, 64), layers.Bidirectional(layers.LSTM(64)), layers.Dense(64, activation='relu'), layers.Dense(1) ]) 其中，Embedding层是用来做word embedding的，把稀疏向量转换成稠密向量；Bidirectional层是一个RNN的双向封装器，用于对序列进行前向和后向计算。 3. 模型编译训练 model.compile(loss=losses.BinaryCrossentropy(from_logits=True), optimizer=optimizers.Adam(1e-4), metrics=['acc']) history = model.fit(train_dataset, epochs=10, validation_data=test_dataset, validation_steps=30) test_loss, test_acc = model.evaluate(test_dataset) print('Test Loss: {}'.format(test_loss)) print('Test Accuracy: {}'.format(test_acc)) 测试集效果如下： 391/391 [==============================] - 109s 280ms/step - loss: 0.4388 - acc: 0.8553 Test Loss: 0.43875041604042053 Test Accuracy: 0.8552799820899963 画个图，看看效果： plt.rcParams['font.size'] = 14 def plot_graphs(history, metric): plt.plot(history.history[metric]) plt.plot(history.history['val_'+metric], '') plt.xlabel("Epochs") plt.ylabel(metric) plt.legend([metric, 'val_'+metric]) plt.show() plot_graphs(history, 'acc') plot_graphs(history, 'loss') 本文参考了如下文章： Text classification with an RNN TensorFlow 2 中文文档 - RNN LSTM 文本分类
PySpark笔记之五：lightGBM调参之PySpark + mmlspark + Hyperopt
Overview1. 加载相关包2. 加载数据3. 定义参数空间和代价函数4. 随机搜索调参和贝叶斯优化调参 Overview 之前的文章中记录了大数据平台上lightGBM分类器的Grid Search调参方法的应用。这次我们继续用lightGBM分类器，看看另外两种常用的调参方法随机搜索Random Search和贝叶斯优化Bayesian Optimization怎么在Spark平台上使用。 1. 加载相关包 import numpy as np import pyspark spark = pyspark.sql.SparkSession.builder.appName("spark lightgbm") \ .master("spark://***.***.***.***:7077") \ .config("spark.jars.packages", "com.microsoft.ml.spark:mmlspark_2.11:0.18.1") \ .config("spark.cores.max", "20") \ .config("spark.driver.memory", "6G") \ .config("spark.executor.memory", "6G") \ .config("spark.executor.cores", "6") \ .getOrCreate() import mmlspark from mmlspark.lightgbm import LightGBMClassifier from pyspark.ml.feature import VectorAssembler from hyperopt import fmin, rand, tpe, hp, space_eval, Trials 其中hyperopt是一款强大的调参工具，实现了随机搜索和贝叶斯优化。 如果没有安装hyperopt,那么先在master节点安装hyperopt： pip3 install hyperopt 2. 加载数据 df_train = spark.read.format("csv") \ .option("inferSchema", "true") \ .option("header", "true") \ .option("sep", ",") \ .load("hdfs://***.***.***.***:39000/yangbingjiao/训练集特征.csv") df_val = spark.read.format("csv") \ .option("inferSchema", "true") \ .option("header", "true") \ .option("sep", ",") \ .load("hdfs://***.***.***.***:39000/yangbingjiao/验证集特征.csv") df_test = spark.read.format("csv") \ .option("inferSchema", "true") \ .option("header", "true") \ .option("sep", ",") \ .load("hdfs://***.***.***.***:39000/yangbingjiao/测试集特征.csv") 处理特征： feature_cols = list(df_train.columns) feature_cols.remove("label") assembler = VectorAssembler(inputCols=feature_cols, outputCol="features") df_train = assembler.transform(df_train) df_val = assembler.transform(df_val) df_test = assembler.transform(df_test) 3. 定义参数空间和代价函数 定义参数搜索空间，以lambdaL1和lambdaL2为例： space = { 'objective':"binary", 'boostingType':'gbdt', 'isUnbalance':True, 'featuresCol':'features', 'labelCol':'label', 'maxBin':60, 'baggingFreq':1, 'baggingSeed':696, 'earlyStoppingRound':20, 'learningRate':0.1, 'maxDepth':3, 'numLeaves':128, 'baggingFraction':0.7, 'featureFraction':0.7, 'minSumHessianInLeaf':0.001, 'numIterations':800, 'verbosity':1, 'lambdaL1': hp.uniform('lambdaL1', 0.0, 200.0), 'lambdaL2': hp.uniform('lambdaL2', 0.0, 200.0), } 定义ks计算函数： import numpy as np from sklearn.metrics import roc_curve, auc def ks_score(y, y_pred, threshold=None): """ 计算KS """ fpr, tpr, thresholds = roc_curve(y, y_pred) if threshold is None: score = np.max(np.abs(tpr - fpr)) else: idx = np.digitize(threshold, thresholds) - 1 score = np.abs(tpr[idx] - fpr[idx]) return score 训练过程记录： def train_evaluate(df_train, df_val, df_test, params): lgb = LightGBMClassifier() model = lgb.fit(df_train, params) val_preds = model.transform(df_val) test_preds = model.transform(df_test) val_prob_list = [row.probability[0] for row in val_preds.select('probability').collect()] val_label_list = [row.label for row in val_preds.select('label').collect()] test_prob_list = [row.probability[0] for row in test_preds.select('probability').collect()] test_label_list = [row.label for row in test_preds.select('label').collect()] val_ks = ks_score(1-np.array(val_label_list), np.array(val_prob_list)) test_ks = ks_score(1-np.array(test_label_list), np.array(test_prob_list)) print("val ks = %f" %(val_ks)) print("test ks = %f" %(test_ks)) print("-------------------------------") return val_ks 定义代价函数： def objective(params): return -train_evaluate(df_train, df_val, df_test, params) 定义代价函数的时候一定要注意，hyperopt会求目标函数的最小值，如果目标函数是loss，那么train_evaluate之前是不用加负号-的；如果我们要求AUC或者KS等最大值，那么train_evaluate之前一定要加上负号-，这样才能得到我们想要的结果。 4. 随机搜索调参和贝叶斯优化调参 随机搜索调参优化： rand_trials = Trials() rand_best = fmin(fn=objective, space=space, algo=rand.suggest, trials=rand_trials, max_evals=5) space_eval(space, rand_best) 输出如下： val ks = 0.380907 test ks = 0.365298 ------------------------------- val ks = 0.372782 test ks = 0.367067 ------------------------------- val ks = 0.370621 test ks = 0.368570 ------------------------------- val ks = 0.383066 test ks = 0.370889 ------------------------------- val ks = 0.378644 test ks = 0.367473 ------------------------------- 100%|██████████| 5/5 [12:04<00:00, 144.83s/it, best loss: -0.38306588065580294] {'baggingFraction': 0.7, 'baggingFreq': 1, 'baggingSeed': 696, 'boostingType': 'gbdt', 'featureFraction': 0.7, 'featuresCol': 'features', 'isUnbalance': True, 'labelCol': 'label', 'lambdaL1': 73.35909229652955, 'lambdaL2': 198.46023902528685, 'learningRate': 0.1, 'maxBin': 60, 'maxDepth': 3, 'minSumHessianInLeaf': 0.001, 'numLeaves': 128, 'objective': 'binary', 'verbosity': 1} 也就是说，随机搜索调参时，KS最大可以达到0.38306588065580294。space_eval这个函数是为了找出最好的那组参数，而Trials类可以记录整个训练过程。 如果要是想用贝叶斯优化，那么这段代码应该是： tpe_trials = Trials() tpe_best = fmin(fn=objective, space=space, algo=tpe.suggest, trials=tpe_trials, max_evals=5) space_eval(space, tpe_best) 输出过程： val ks = 0.392768 test ks = 0.373026 ------------------------------- val ks = 0.383776 test ks = 0.367786 ------------------------------- val ks = 0.385806 test ks = 0.370686 ------------------------------- val ks = 0.375033 test ks = 0.370436 ------------------------------- val ks = 0.388076 test ks = 0.371288 ------------------------------- 100%|██████████| 5/5 [10:51<00:00, 130.20s/it, best loss: -0.39276827239870765] {'baggingFraction': 0.7, 'baggingFreq': 1, 'baggingSeed': 696, 'boostingType': 'gbdt', 'featureFraction': 0.7, 'featuresCol': 'features', 'isUnbalance': True, 'labelCol': 'label', 'lambdaL1': 52.782517457829606, 'lambdaL2': 20.799935111367063, 'learningRate': 0.1, 'maxBin': 60, 'maxDepth': 3, 'minSumHessianInLeaf': 0.001, 'numLeaves': 128, 'objective': 'binary', 'verbosity': 1} 使用贝叶斯优化时，KS可以达到0.39276827239870765。可以看到，贝叶斯优化还是要好于随机搜索算法的。 我们在拿到最优参数后，就可以训练最佳的模型了。 很不幸，mmlspark当中实现的lightGBM算法，当前很可能是无法复现的，也就是说少了一个控制随机性的seed。我们可以看到，LightGBMClassifier这个类，接收的参数当中，只有一个baggingSeed，用来控制baggingFraction时每次抽取哪些样本；而featureFraction显然也需要一个参数来控制每次抽取哪些特征。当然，原生的lightGBM当中的seed参数是可以控制这两个随机性的。这个问题我已经在GitHub上进行反馈了，有兴趣可以去该页面留言 lightgbm run twice with the same parameters, but got different result in validation #564，催微软更新，或者自己贡献代码。我会继续关注这个问题。 本文参考了： Parameter Tuning with Hyperopt Introduction: Bayesian Optimization using Hyperopt An Introductory Example of Bayesian Optimization in Python with Hyperopt Hyperopt tutorial for Optimizing Neural Networks’ Hyperparameters Hyperopt
PySpark笔记之四：lightGBM调参之PySpark + mmlspark + Grid Search
Overview1. 加载PySpark和mmlspark相关包2.加载数据3.建模4.预测测试集 Overview 上一篇文章PySpark笔记之三：lightGBM调参之PySpark + Grid Search，我们记录了分别用PySpark中自带的CrossValidator和更通用的生成多个分类器同时执行训练预测的方式选取最好的模型。其中CrossValidator并不能得到验证集上最佳的分类器，而是得到训练集上最佳的效果。而mmlspark当中却有更为简单的方式，既可以得到验证集上最佳的效果，也可以方便地记录我们每一组参数对应的结果，是一种很好的方式。 1. 加载PySpark和mmlspark相关包 import numpy as np import pyspark spark = pyspark.sql.SparkSession.builder.appName("spark lightgbm") \ .master("spark://***.***.***.***:7077") \ .config("spark.jars.packages", "com.microsoft.ml.spark:mmlspark_2.11:0.18.1") \ .config("spark.cores.max", "20") \ .config("spark.driver.memory", "6G") \ .config("spark.executor.memory", "6G") \ .config("spark.executor.cores", "6") \ .getOrCreate() from pyspark.ml.feature import VectorAssembler from pyspark.ml.evaluation import BinaryClassificationEvaluator import mmlspark from mmlspark.lightgbm import LightGBMClassifier from mmlspark.automl import * from mmlspark.train import TrainClassifier, ComputeModelStatistics 2.加载数据 df_train = spark.read.format("csv") \ .option("inferSchema", "true") \ .option("header", "true") \ .option("sep", ",") \ .load("hdfs://***.***.***.***:39000/young/训练集特征.csv") df_val = spark.read.format("csv") \ .option("inferSchema", "true") \ .option("header", "true") \ .option("sep", ",") \ .load("hdfs://***.***.***.***:39000/young/验证集特征.csv") df_test = spark.read.format("csv") \ .option("inferSchema", "true") \ .option("header", "true") \ .option("sep", ",") \ .load("hdfs://***.***.***.***:39000/young/测试集特征.csv") 处理训练集、验证集和测试集数据： feature_cols = list(df_train.columns) feature_cols.remove("label") assembler = VectorAssembler(inputCols=feature_cols, outputCol="features") df_train = assembler.transform(df_train) df_val = assembler.transform(df_val) df_test = assembler.transform(df_test) 3.建模 lgb = LightGBMClassifier( objective="binary", boostingType='gbdt', isUnbalance=True, featuresCol='features', labelCol='label', maxBin=60, baggingFreq=1, baggingSeed=696, earlyStoppingRound=20, learningRate=0.1, #lambdaL1=1.0, #lambdaL2=45.0, maxDepth=3, numLeaves=128, baggingFraction=0.7, featureFraction=0.7, minSumHessianInLeaf=0.001, numIterations=800, verbosity=1 ) 以lambdaL1和lambdaL2为例，设置4组不同的参数，这里可以自己记录每组参数，方便和后面各模型效果对应，就不再赘述了： lightGBMs = list() for lambdaL1 in list(np.arange(1.0, 3.0, 1.0)): for lambdaL2 in list(np.arange(1.0, 3.0, 1.0)): lightGBMs.append(lgb.setLambdaL1(lambdaL1).setLambdaL2(lambdaL2)) 当然，这里可以用这种方式来设置参数，更方便： import itertools lightGBMs = list() params = itertools.product([1.0, 2.0], [1.0, 2.0]) for param in params: lightGBMs.append(lgb.setLambdaL1(param[0]).setLambdaL2(param[1])) 利用mmlspark.train模块当中的TrainClassifier类训练模型： lgb_models = [TrainClassifier(model=lgb, labelCol="label").fit(df_train) for lgb in lightGBMs] 利用mmlspark.automl当中的FindBestModel类，寻找在验证集上效果最好的模型： best_model = FindBestModel(evaluationMetric='AUC', models=lgb_models).fit(df_val) 我们可以看看最好的模型效果： best_model.getBestModelMetrics().collect() 结果如下，是一个长度为1的list： [Row(evaluation_type='Classification', confusion_matrix=DenseMatrix(2, 2, [6647.0, 578.0, 2774.0, 1219.0], False), accuracy=0.7011945088251025, precision=0.30528424743300775, recall=0.6783528102392877, AUC=0.7554637200060459)] 也可以看看所有4个模型在验证集上的效果： best_model.getAllModelMetrics().collect() 结果是个list，长度为4： [Row(model_name='TrainClassifier_64edf29511fb', metric=0.7544038268737058, parameters='featuresCol: TrainClassifier_31b56e2ce33a_features, labelCol: label, predictionCol: prediction, probabilityCol: probability, rawPredictionCol: rawPrediction'), Row(model_name='TrainClassifier_b09abcd6c7e5', metric=0.7541219526558816, parameters='featuresCol: TrainClassifier_9c85465a6f6d_features, labelCol: label, predictionCol: prediction, probabilityCol: probability, rawPredictionCol: rawPrediction'), Row(model_name='TrainClassifier_5a1bd3015154', metric=0.7528073862858738, parameters='featuresCol: TrainClassifier_ba6c57087959_features, labelCol: label, predictionCol: prediction, probabilityCol: probability, rawPredictionCol: rawPrediction'), Row(model_name='TrainClassifier_323901015bd2', metric=0.7554575473623404, parameters='featuresCol: TrainClassifier_59842f00f628_features, labelCol: label, predictionCol: prediction, probabilityCol: probability, rawPredictionCol: rawPrediction')] 4.预测测试集 predictions = bestModel.transform(df_test) metrics = ComputeModelStatistics().transform(predictions) print("Best model's AUC on test set = " + "{0:.2f}%".format(metrics.first()["AUC"] * 100)) 我们可以看看metrics里面都是什么： metrics.collect() 结果也是一个list，包含混淆矩阵，accuracy,precision和recall： [Row(evaluation_type='Classification', confusion_matrix=DenseMatrix(2, 2, [11323.0, 1777.0, 83.0, 115.0], False), accuracy=0.8601293427583095, precision=0.5808080808080808, recall=0.060782241014799156, AUC=0.7411781436942614)] 总体来说，这种方式来进行分布式训练，比PySpark自带的api更方便一些，推荐mmlspark方式。 本文参考了以下文章： Simplifying Machine Learning Pipelines with mmlspark
PySpark笔记之三：lightGBM调参之PySpark + Grid Search
Overview1. 引入依赖包2. 加载数据3. 模型构建4. 交叉验证选择模型5. 官方调参探讨 Overview 在实际机器学习工作当中，调参是我们一个重要的内容。PySpark当中就实现了一个最常用的调参方法Grid Search，我们结合lightGBM使用一下PySpark的调参。这个程序需要安装的依赖的安装方式，可以参考上一篇博客。 1. 引入依赖包 import numpy as np import pyspark spark = pyspark.sql.SparkSession.builder.appName("spark lightgbm") \ .master("spark://***.***.***.***:7077") \ .config("spark.jars.packages", "com.microsoft.ml.spark:mmlspark_2.11:0.18.1") \ .config("spark.cores.max", "20") \ .config("spark.driver.memory", "6G") \ .config("spark.executor.memory", "6G") \ .config("spark.executor.cores", "6") \ .getOrCreate() import mmlspark from mmlspark.lightgbm import LightGBMClassifier from pyspark.ml.feature import VectorAssembler from pyspark.ml.tuning import CrossValidator, ParamGridBuilder from pyspark.ml.evaluation import BinaryClassificationEvaluator from pyspark.ml import Pipeline 其中，pyspark.ml.tuning.ParamGridBuilder就是用以实现Grid Search的包。 2. 加载数据 df_train = spark.read.format("csv") \ .option("inferSchema", "true") \ .option("header", "true") \ .option("sep", ",") \ .load("hdfs://***.***.***.***:39000/young/训练集特征.csv") df_val = spark.read.format("csv") \ .option("inferSchema", "true") \ .option("header", "true") \ .option("sep", ",") \ .load("hdfs://***.***.***.***:39000/young/验证集特征.csv") df_test = spark.read.format("csv") \ .option("inferSchema", "true") \ .option("header", "true") \ .option("sep", ",") \ .load("hdfs://***.***.***.***:39000/young/测试集特征.csv") 处理训练集特征： feature_cols = list(df_train.columns) feature_cols.remove("label") # 从列名当中删除label才是真正的特征列表 assembler = VectorAssembler(inputCols=feature_cols, outputCol="features") df_train = assembler.transform(df_train) 3. 模型构建 lgb = LightGBMClassifier( objective="binary", boostingType='gbdt', isUnbalance=True, featuresCol='features', labelCol='label', maxBin=60, baggingFreq=1, baggingSeed=696, earlyStoppingRound=30, learningRate=0.1, # lambdaL1=1.0, # lambdaL2=45.0, maxDepth=3, numLeaves=128, baggingFraction=0.7, featureFraction=0.7, minSumHessianInLeaf=0.001, numIterations=800, verbosity=1 ) 设置Grid Search参数组： paramGrid = ParamGridBuilder() \ .addGrid(lgb.lambdaL1, list(np.arange(1.0, 3.0, 1.0))) \ .addGrid(lgb.lambdaL2, list(np.arange(1.0, 4.0, 1.0))) \ .build() 设置完成之后，我们可以看一下参数都是哪些： for param in paramGrid: print(param.values()) 可以看到共6组参数，左边为lambdaL1，右边为lambdaL2： dict_values([1.0, 1.0]) dict_values([1.0, 2.0]) dict_values([1.0, 3.0]) dict_values([2.0, 1.0]) dict_values([2.0, 2.0]) dict_values([2.0, 3.0]) 4. 交叉验证选择模型 官方提供了两种模型选择的方式：CrossValidator和TrainValidationSplit，可以参考官方文档。CrossValidator和TrainValidationSplit的区别在于：CrossValidator会每次选取一部分训练集建模，去预测另外一部分训练集，这样会有K个预测的分数（K折交叉验证），最后模型的预测分数为K个分数的平均；而TrainValidationSplit则只会训练预测一次。这里，我们试着给出一个CrossValidator的例子。 evaluator = BinaryClassificationEvaluator(labelCol="label", metricName="areaUnderROC") cross_vallidator = CrossValidator(estimator=lgb, estimatorParamMaps=paramGrid, evaluator=evaluator, numFolds=3) model = cross_vallidator.fit(df_train) 最后，我们可以得到最好的模型，并看最好的模型的特征重要性： model.bestModel model.bestModel.getFeatureImportances() 但是，官方api查看每个结果对应的超参数却是非常不友好，我们只好自己想办法，这里我们参考了这篇博客LightGBM Hyper Parameters Tuning in Spark： def params_extract(model): """ function extact hyperparameter information from a CrossValidatorModel input: a CrossValidatorModel instance, model fit by CrossValidator in pyspark.ml.tuning output: a dictionary with key(hyperparameters setting), value(evaluator's metrics, r2, auc,...) """ length = len(model.avgMetrics) res = {} for i in range(length): s = "" paraDict = model.extractParamMap()[model.estimatorParamMaps][i] for j in paraDict.keys(): s += str(j).split("__")[1] + " " s += str(paraDict[j]) + " " res[s.strip()] = model.avgMetrics[i] return {k: v for k, v in sorted(res.items(), key=lambda item: item[1])} 我们试一下： params_extract(model) 输出如下： {'lambdaL1 1.0 lambdaL2 2.0': 0.7300710699287217, 'lambdaL1 1.0 lambdaL2 1.0': 0.7307078416518147, 'lambdaL1 1.0 lambdaL2 3.0': 0.7310860740685388, 'lambdaL1 2.0 lambdaL2 1.0': 0.7312930665848859, 'lambdaL1 2.0 lambdaL2 2.0': 0.7317563737747359, 'lambdaL1 2.0 lambdaL2 3.0': 0.7327463775422981} 5. 官方调参探讨 我们通过CrossValidator可以获得最佳的模型，但是会有一个问题：这个最佳是拟合训练集的最佳，而不是我们给出的验证集的最佳；即使是TrainValidationSplit，我们也不能自定义验证集并传入，只能随机选择验证集。这样对于那些样本时间先后顺序不敏感的数据是影响不大的，比如图像等，但是对于交易类数据，我们希望可以根据时间先后顺序自定义训练集，验证集和测试集，并且根据验证集的效果来确定最佳参数和模型。因此，我们就需要换种方式达到目的。 import numpy as np import pyspark spark = pyspark.sql.SparkSession.builder.appName("spark lightgbm") \ .master("spark://***.***.***.***:7077") \ .config("spark.jars.packages", "com.microsoft.ml.spark:mmlspark_2.11:0.18.1") \ .config("spark.cores.max", "20") \ .config("spark.driver.memory", "6G") \ .config("spark.executor.memory", "6G") \ .config("spark.executor.cores", "6") \ .getOrCreate() import mmlspark from mmlspark.lightgbm import LightGBMClassifier from pyspark.ml.feature import VectorAssembler from pyspark.ml.tuning import CrossValidator, ParamGridBuilder from pyspark.ml.evaluation import BinaryClassificationEvaluator # 加载数据 df_train = spark.read.format("csv") \ .option("inferSchema", "true") \ .option("header", "true") \ .option("sep", ",") \ .load("hdfs://***.***.***.***:39000/young/训练集特征.csv") df_val = spark.read.format("csv") \ .option("inferSchema", "true") \ .option("header", "true") \ .option("sep", ",") \ .load("hdfs://***.***.***.***:39000/young/验证集特征.csv") df_test = spark.read.format("csv") \ .option("inferSchema", "true") \ .option("header", "true") \ .option("sep", ",") \ .load("hdfs://***.***.***.***:39000/young/测试集特征.csv") feature_cols = list(df_train.columns) feature_cols.remove("label") assembler = VectorAssembler(inputCols=feature_cols, outputCol="features") df_train = assembler.transform(df_train) df_val = assembler.transform(df_val) df_test = assembler.transform(df_test) lgb = LightGBMClassifier( objective="binary", boostingType='gbdt', isUnbalance=True, featuresCol='features', labelCol='label', maxBin=60, baggingFreq=1, baggingSeed=696, earlyStoppingRound=20, learningRate=0.1, #lambdaL1=1.0, #lambdaL2=45.0, maxDepth=3, numLeaves=128, baggingFraction=0.7, featureFraction=0.7, minSumHessianInLeaf=0.001, numIterations=800, verbosity=1 ) lightGBMs = list() for lambdaL1 in list(np.arange(1.0, 3.0, 1.0)): for lambdaL2 in list(np.arange(1.0, 4.0, 1.0)): lightGBMs.append(lgb.setLambdaL1(lambdaL1).setLambdaL2(lambdaL2)) evaluator = BinaryClassificationEvaluator(rawPredictionCol="rawPrediction", metricName="areaUnderROC") metrics = [] models = [] # 选择验证集效果最好的模型 for learner in lightGBMs: model = learner.fit(df_train) models.append(model) scoredData = model.transform(df_val) metrics.append(evaluator.evaluate(scoredData)) best_metric = max(metrics) best_model = models[metrics.index(best_metric)] # 得到测试集上AUC scored_test = best_model.transform(df_test) print(evaluator.evaluate(scored_test)) 当然，用这种方式的话，我们就可以很方便地记录每组参数对应的结果了，在评价验证集效果时，也可以用KS或者Gini等指标了，这里就不再赘述。 本文主要参考了：How to build machine learning model at large scale with Apache Spark and LightGBM for credit card fraud detection? Simplifying Machine Learning Pipelines with mmlspark
Overview1. 安装mmlspark2. 引入依赖包3. 加载数据4. 模型构建 Overview 就目前的PySpark版本2.4.5来说，虽有pyspark.ml这个模块可以进行机器学习，但是都是一些工业界不太常用的算法，而XGBoost和LightGBM这样的常用算法还没有集成。幸好微软前几年发布了mmlspark这个包，其中包含了深度学习和LightGBM等算法，可以和PySpark无缝对接。下面我们看看怎么用PySpark和mmlspark来运行LightGBM。 1. 安装mmlspark 首先，我们默认已经安装好了PySpark，如果没有安装，那么安装命令如下： pip3 install pyspark 然后，安装mmlspark的方式如下，命令行输入： pyspark --packages com.microsoft.ml.spark:mmlspark_2.11:0.18.1 或者直接在jupyter notebook中，使用PySpark时，这样启动： import pyspark spark = pyspark.sql.SparkSession.builder.appName("MyApp") \ .config("spark.jars.packages", "com.microsoft.ml.spark:mmlspark_2.11:1.0.0-rc1") \ .config("spark.jars.repositories", "https://mmlspark.azureedge.net/maven") \ .getOrCreate() import mmlspark 这个包比较大，第一次安装需要较长时间。我们服务器上maven仓库当中，mmlspark版本是0.18.1，而这个版本训练时会导致一个bug，下面会提到。 详细安装方式可参考官方文档：Azure / mmlspark，最新的mmlspark版本是1.0.0-rc1。 2. 引入依赖包 import numpy as np import pyspark spark = pyspark.sql.SparkSession.builder.appName("spark lightgbm") \ .master("spark://***.***.***.***:7077") \ .config("spark.jars.packages", "com.microsoft.ml.spark:mmlspark_2.11:0.18.1") \ .config("spark.cores.max", "20") \ .config("spark.driver.memory", "6G") \ .config("spark.executor.memory", "6G") \ .config("spark.executor.cores", "6") \ .getOrCreate() import mmlspark from mmlspark.lightgbm import LightGBMClassifier from pyspark.ml.feature import VectorAssembler # from pyspark.ml.tuning import CrossValidator, ParamGridBuilder from pyspark.ml.evaluation import BinaryClassificationEvaluator from pyspark.ml import Pipeline 其中，VectorAssembler这个类，可以将特征转换成一个向量，作为分类器的输入。BinaryClassificationEvaluator可以评价预测效果。Pipeline相当于Spark的流水线，将各个步骤连接在一起后，一起由Spark运行。 3. 加载数据 df_train = spark.read.format("csv") \ .option("inferSchema", "true") \ .option("header", "true") \ .option("sep", ",") \ .load("hdfs://***.***.***.***:39000/young/训练集特征.csv") df_val = spark.read.format("csv") \ .option("inferSchema", "true") \ .option("header", "true") \ .option("sep", ",") \ .load("hdfs://***.***.***.***:39000/young/验证集特征.csv") df_test = spark.read.format("csv") \ .option("inferSchema", "true") \ .option("header", "true") \ .option("sep", ",") \ .load("hdfs://***.***.***.***:39000/young/测试集特征.csv") 首先，要保证Hadoop上有我们需要的数据集文件。本地文件上传到Hadoop的命令很简单： hdfs dfs -copyFromLocal 训练集特征.csv 验证集特征.csv 测试集特征.csv /young 处理训练集特征： feature_cols = list(df_train.columns) feature_cols.remove("label") # 从列名当中删除label才是真正的特征列表 assembler = VectorAssembler(inputCols=feature_cols, outputCol="features") VectorAssembler之后，原数据DataFrame当中会多出一列，列名为features，这个字段内容其实就是一个行向量，里面是该样本的每一个特征。 4. 模型构建 lgb = LightGBMClassifier( objective="binary", boostingType='gbdt', isUnbalance=True, featuresCol='features', labelCol='label', maxBin=60, baggingFreq=1, baggingSeed=696, earlyStoppingRound=30, learningRate=0.1, lambdaL1=1.0, lambdaL2=45.0, maxDepth=3, numLeaves=128, baggingFraction=0.7, featureFraction=0.7, # minSumHessianInLeaf=1, numIterations=800, verbosity=30 ) 其中，minSumHessianInLeaf这个参数，在设置为整型数值时，会抛ClassCastException异常， Py4JJavaError: An error occurred while calling o1592.w. : java.lang.ClassCastException: java.lang.Integer cannot be cast to java.lang.Double at scala.runtime.BoxesRunTime.unboxToDouble(BoxesRunTime.java:114) at org.apache.spark.ml.param.DoubleParam.w(params.scala:330) at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method) at sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62) at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43) at java.lang.reflect.Method.invoke(Method.java:498) at py4j.reflection.MethodInvoker.invoke(MethodInvoker.java:244) at py4j.reflection.ReflectionEngine.invoke(ReflectionEngine.java:357) at py4j.Gateway.invoke(Gateway.java:282) at py4j.commands.AbstractCommand.invokeMethod(AbstractCommand.java:132) at py4j.commands.CallCommand.execute(CallCommand.java:79) at py4j.GatewayConnection.run(GatewayConnection.java:238) at java.lang.Thread.run(Thread.java:748) 这是一个bug，在最新版本当中，这个bug应该已经改过来了。不过，对0.18.1这样比较老的版本，我们可以将这个变量设置为double或者float类型，例如1写成1.0，那么这个异常就不会出现了。 装载各个阶段到Pipeline流水线中，执行训练： stages = [assembler, lgb] pipeline_model = Pipeline(stages=stages) model = pipeline_model.fit(df_train) 预测三个数据集: train_preds = model.transform(df_train) val_preds = model.transform(df_val) val_preds = model.transform(df_val) 我们看看模型训练效果： binaryEvaluator = BinaryClassificationEvaluator() print ("Train AUC: " + str(binaryEvaluator.evaluate(train_preds, {binaryEvaluator.metricName: "areaUnderROC"}))) print ("Val AUC: " + str(binaryEvaluator.evaluate(val_preds, {binaryEvaluator.metricName: "areaUnderROC"}))) print ("Test AUC: " + str(binaryEvaluator.evaluate(test_preds, {binaryEvaluator.metricName: "areaUnderROC"}))) 输出三个AUC： Train AUC: 0.8236477163798177 Val AUC: 0.7548691969544107 Test AUC: 0.749396806843621 当然，我们可以把预测概率结果和真实label取出来，方便进行计算其他自定义指标，例如KS等。 train_prob_list = [row.probability[0] for row in train_preds.select('probability').collect()] train_label_list = [row.label for row in train_preds.select('label').collect()] val_prob_list = [row.probability[0] for row in val_preds.select('probability').collect()] val_label_list = [row.label for row in val_preds.select('label').collect()] test_prob_list = [row.probability[0] for row in test_preds.select('probability').collect()] test_label_list = [row.label for row in test_preds.select('label').collect()] DataFrame在经过一系列transform之后，会多出4列，分别为features|rawPrediction|probability|prediction|，rawPrediction是树模型最后的一对儿得分，和为0；在经过sigmoid之后，得到probability里的一对儿概率值，其和为1，分别表示模型判定该样本为两个分类的可能性；而prediction则是模型预测的样本类别。 本文参考了以下文章： 在集群上训练一个机器学习模型（以lightGBM为例） 利用pyspark.ml训练lightgbm模型的流程 Get Started with PySpark and LightGBM Build XGBoost / LightGBM models on large datasets — what are the possible solutions?