NVTabular demo on RecSys2020 Challenge
Overview
NVTabular is a feature engineering and preprocessing library for tabular data designed to quickly and easily manipulate terabyte scale datasets used to train deep learning based recommender systems. It provides a high level abstraction to simplify code and accelerates computation on the GPU using the RAPIDS cuDF library.
RecSys2020 Challenge
The RecSys conference is the leading data science conference for recommender systems and organizes an annual competiton in recommender systems. The RecSys Challenge 2020, hosted by Twitter, was about predicting interactions of ~200 mio. tweet-user pairs. NVIDIA’s team scored 1st place. The team explained the solution in the blogpost and published the code on github.
Downloading the dataset
The dataset has to be downloaded from the original source, provided by Twitter. You need to create an account on the RecSys Challenge 2020 website. Twitter needs to (manually) approve your account, which can take a few days. After your account is approved, you can download the data here. We will use only the training.tsv file, as we cannot make submissions anymore.
Learning objectives
This notebook covers the end-2-end pipeline, from loading the original .tsv file to training the models, with NVTabular, cuDF, dask and XGBoost. We demonstrate multi-GPU support for NVTabular and new nvt.ops, implemented based on the success of our RecSys2020 solution. 1. NVTabular to preprocess the original .tsv file. 2. dask_cudf to split the preprocessed data into a training and
validation set. 3. NVTabular to create additional features with only ~70 lines of code. 4. dask_cudf / XGBoost on GPU to train our model.
Getting Started
[1]:
# External Dependencies
import time
import glob
import gc
import cupy as cp # CuPy is an implementation of NumPy-compatible multi-dimensional array on GPU
import cudf # cuDF is an implementation of Pandas-like Dataframe on GPU
import rmm # library for pre-allocating memory on GPU
import dask # dask is an open-source library to nateively scale Python on multiple workers/nodes
import dask_cudf # dask_cudf uses dask to scale cuDF dataframes on multiple workers/nodes
import numpy as np
# NVTabular is the core library, we will use here for feature engineering/preprocessing on GPU
import nvtabular as nvt
import xgboost as xgb
# More dask / dask_cluster related libraries to scale NVTabular
from dask_cuda import LocalCUDACluster
from dask.distributed import Client
from dask.distributed import wait
from dask.utils import parse_bytes
from dask.delayed import delayed
from nvtabular.utils import device_mem_size
Let’s have a short look on the libary versions and setup, we will use.
[2]:
cudf.__version__
[2]:
'0+untagged.1.gfa8e9fb'
[3]:
cp.__version__
[3]:
'7.8.0'
[4]:
xgb.__version__
[4]:
'1.2.0-SNAPSHOT'
[5]:
dask.__version__
[5]:
'2.24.0'
We ran this example with 2x Quadro GV100 GPUs with each having 32GB of GPU memory.
[6]:
!nvidia-smi
Wed Nov 11 15:39:04 2020
+-----------------------------------------------------------------------------+
| NVIDIA-SMI 450.66 Driver Version: 450.66 CUDA Version: 11.0 |
|-------------------------------+----------------------+----------------------+
| GPU Name Persistence-M| Bus-Id Disp.A | Volatile Uncorr. ECC |
| Fan Temp Perf Pwr:Usage/Cap| Memory-Usage | GPU-Util Compute M. |
| | | MIG M. |
|===============================+======================+======================|
| 0 Quadro GV100 Off | 00000000:15:00.0 Off | Off |
| 34% 45C P2 26W / 250W | 13MiB / 32508MiB | 0% Default |
| | | N/A |
+-------------------------------+----------------------+----------------------+
| 1 Quadro GV100 Off | 00000000:2D:00.0 On | Off |
| 39% 51C P0 34W / 250W | 1180MiB / 32499MiB | 0% Default |
| | | N/A |
+-------------------------------+----------------------+----------------------+
+-----------------------------------------------------------------------------+
| Processes: |
| GPU GI CI PID Type Process name GPU Memory |
| ID ID Usage |
|=============================================================================|
+-----------------------------------------------------------------------------+
[7]:
!free
total used free shared buff/cache available
Mem: 48020044 5775436 37876780 464524 4367828 41193124
Swap: 1003516 1003444 72
[8]:
time_total_start = time.time()
We define our base directory, containing the data.
[9]:
BASE_DIR = '/raid/data/recsys2020/'
First, we initalize our local cuda cluster.
[10]:
cluster = LocalCUDACluster(
protocol="tcp"
)
client = Client(cluster)
client
[10]:
Client
|
Cluster
|
Preparing our dataset
The original data format had multiple inefficiencies, resulting in requiring more disk space and memory: 1. The file format is .tsv, an uncompressed, column-separated format. The parquet file format stores tabular data in a compressed column-oriented format. This saves a significant amount of disk space which results in fewer i/o operations and faster execution. 2. Some categorical features, such as tweet_id, user_id are hashed to String values
(e.g. cfcd208495d565ef66e7dff9f98764da). These long Strings require significant amount of disk space/memory. We can encode the Categories as Integer values using Categorify. Representing the String cfcd208495d565ef66e7dff9f98764da as an Integer 0 can save up 90% in memory. Although other categorical features are no long hashes, they are still Strings (e.g. tweet_type) and we will represent them as Integers, as well. 3. In our experiments, the text_tokens were not a significant
feature and we will drop the column before we split the data.
First, we define the data schema: - features are the column names in the original .tsv file. The .tsv file has no header and we need to specify the names - cat_names are the categorical features, we want to use. - cont_names are the numerical features, we want to use. - label_name contains the target column.
[11]:
features = [
'text_tokens', ###############
'hashtags', #Tweet Features
'tweet_id', #
'media', #
'links', #
'domains', #
'tweet_type', #
'language', #
'timestamp', ###############
'a_user_id', ###########################
'a_follower_count', #Engaged With User Features
'a_following_count', #
'a_is_verified', #
'a_account_creation', ###########################
'b_user_id', #######################
'b_follower_count', #Engaging User Features
'b_following_count', #
'b_is_verified', #
'b_account_creation', #######################
'b_follows_a', #################### Engagement Features
'reply', #Target Reply
'retweet', #Target Retweet
'retweet_comment',#Target Retweet with comment
'like', #Target Like
####################
]
cat_names = [
'hashtags',
'tweet_id',
'media',
'links',
'domains',
'tweet_type',
'language',
'a_user_id',
'a_is_verified',
'b_user_id',
'b_is_verified',
'b_follows_a',
]
cont_names = [
'timestamp',
'a_follower_count',
'a_following_count',
'a_account_creation',
'b_follower_count',
'b_following_count',
'b_account_creation'
]
label_name = [
'reply',
'retweet',
'retweet_comment',
'like'
]
We initialize our NVTabular workflow.
[12]:
proc = nvt.Workflow(cat_names=cat_names,
cont_names=cont_names,
label_name=label_name)
We define two helper function, we apply in our NVTabular workflow: 1. splitmedia2 splits the entries in media by \t and keeps only the first two values (if available), 2. count_token counts the number of token in a column (e.g. how many hashtags are in a tweet),
[13]:
def splitmedia(col):
if col.shape[0] == 0:
return(col)
else:
return(col.str.split('\t')[0].fillna('') + '_' + col.str.split('\t')[1].fillna(''))
def count_token(col,token):
not_null = col.isnull()==0
return ((col.str.count(token)+1)*not_null).fillna(0)
We initialize a nvt.Dataset. The engine is csv as the .tsv file has a similar structure. The .tsv file uses the special character \x01 to separate columns. There is no header in the file and we define column names with the parameter names.
[14]:
trains_itrs = nvt.Dataset(BASE_DIR + 'training.tsv',
header=None,
names=features,
engine='csv',
sep='\x01',
part_size='1GB')
NVTabular ops can be registered in two different ways to a NVTabular workflow: Feature or Preprocess. Features are executed before Preprocess ``ops``. * The functions add_cat_feature, add_cont_feature or add_feature register an op as feature * The functions add_cat_preprocess, add_cont_preprocess or add_preprocess register an op as preprocess
We count the number of tokens in the columns hashtags, domains, links. `LambdaOp <https://nvidia.github.io/NVTabular/main/api/ops/lambdaop.html>`__ is a special NVTabular operator, which provides an easy way to apply a row-level, user-defined functions to the dataframe.
[15]:
proc.add_feature(
nvt.ops.LambdaOp(
op_name='count_t',
f=lambda col, gdf: count_token(col,'\t'),
columns=['hashtags', 'domains', 'links'],
replace=False
)
)
We apply splitmedia function to split the media.
[16]:
proc.add_feature(
nvt.ops.LambdaOp(
op_name='splitmedia',
f=lambda col, gdf: splitmedia(col),
columns=['media'],
replace=True
)
)
We fill in na/missing values in target columns + hashtags, domains, links.
[17]:
proc.add_feature(
nvt.ops.FillMissing(columns=label_name + ['hashtags', 'domains', 'links'])
)
We encode columns as a small, continuous integer to save memory. Some categorical columns contain long hashes of type String as values to preserve the privacy of the users (e.g. userId, language, etc.). Long hashes of type String requires significant amount of memory to store. We encode/map the Strings to continuous Integer to save significant memory.
[18]:
proc.add_preprocess([
nvt.ops.Categorify(
columns=['media', 'language', 'tweet_type', 'tweet_id',
'a_user_id', 'b_user_id', 'hashtags', 'domains', 'links']
)
])
We extract weekday from the timestamp column.
[19]:
proc.add_preprocess(
nvt.ops.LambdaOp(
op_name = 'wd',
f = lambda col, gdf: cudf.to_datetime(col, unit='s').dt.weekday,
columns = ['timestamp'],
replace=False
)
)
We define the output datatypes for continuous columns to save memory. We can define the output datatypes as a dict and parse it to the apply function.
[20]:
dict_dtypes = {}
for col in label_name + ['media', 'language', 'tweet_type', 'tweet_id',
'a_user_id', 'b_user_id', 'hashtags', 'domains',
'links', 'timestamp', 'a_follower_count',
'a_following_count', 'a_account_creation',
'b_follower_count', 'b_following_count', 'b_account_creation']:
dict_dtypes[col] = np.uint32
We apply the workflow and save the output to preprocess/.
[21]:
%%time
time_preproc_start = time.time()
proc.apply(trains_itrs, record_stats=True, output_path=BASE_DIR + 'preprocess/', dtypes=dict_dtypes)
time_preproc = time.time()-time_preproc_start
/opt/conda/envs/rapids/lib/python3.7/site-packages/cudf/core/column/string.py:1992: UserWarning: `expand` parameter defatults to True.
warnings.warn("`expand` parameter defatults to True.")
CPU times: user 2min 43s, sys: 1min 29s, total: 4min 13s
Wall time: 5min 37s
We can take a look in the output folder.
[22]:
!ls $BASE_DIR/preprocess
_metadata part.2.parquet part.31.parquet part.43.parquet
part.0.parquet part.20.parquet part.32.parquet part.44.parquet
part.1.parquet part.21.parquet part.33.parquet part.45.parquet
part.10.parquet part.22.parquet part.34.parquet part.46.parquet
part.11.parquet part.23.parquet part.35.parquet part.47.parquet
part.12.parquet part.24.parquet part.36.parquet part.48.parquet
part.13.parquet part.25.parquet part.37.parquet part.5.parquet
part.14.parquet part.26.parquet part.38.parquet part.6.parquet
part.15.parquet part.27.parquet part.39.parquet part.7.parquet
part.16.parquet part.28.parquet part.4.parquet part.8.parquet
part.17.parquet part.29.parquet part.40.parquet part.9.parquet
part.18.parquet part.3.parquet part.41.parquet
part.19.parquet part.30.parquet part.42.parquet
Splitting dataset into training and test
We split the training data by time into a train and validation set. The first 5 days are train and the last 2 days are for validation. We use the weekday for it. The first day of the dataset is a Thursday (weekday id = 3) and the last day is Wednesday (weekday id = 2). Therefore, we split the weekday ids 1 and 2 into the validation set.
[23]:
%%time
time_split_start = time.time()
df = dask_cudf.read_parquet(BASE_DIR + 'preprocess/*.parquet')
if 'text_tokens' in list(df.columns):
df = df.drop('text_tokens', axis=1)
VALID_DOW = [1, 2]
valid = df[df['timestamp_wd'].isin(VALID_DOW)].reset_index(drop=True)
train = df[~df['timestamp_wd'].isin(VALID_DOW)].reset_index(drop=True)
train = train.sort_values(["b_user_id", "timestamp"]).reset_index(drop=True)
valid = valid.sort_values(["b_user_id", "timestamp"]).reset_index(drop=True)
train.to_parquet(BASE_DIR + 'nv_train/')
valid.to_parquet(BASE_DIR + 'nv_valid/')
time_split = time.time()-time_split_start
del train; del valid
gc.collect()
CPU times: user 11.2 s, sys: 26.7 ms, total: 11.2 s
Wall time: 2min 2s
[23]:
458
Feature Engineering
Now, we can apply the actual feature engineering. First, we define the data schema, again.
[24]:
!nvidia-smi
Wed Nov 11 15:46:49 2020
+-----------------------------------------------------------------------------+
| NVIDIA-SMI 450.66 Driver Version: 450.66 CUDA Version: 11.0 |
|-------------------------------+----------------------+----------------------+
| GPU Name Persistence-M| Bus-Id Disp.A | Volatile Uncorr. ECC |
| Fan Temp Perf Pwr:Usage/Cap| Memory-Usage | GPU-Util Compute M. |
| | | MIG M. |
|===============================+======================+======================|
| 0 Quadro GV100 Off | 00000000:15:00.0 Off | Off |
| 42% 54C P2 49W / 250W | 1477MiB / 32508MiB | 0% Default |
| | | N/A |
+-------------------------------+----------------------+----------------------+
| 1 Quadro GV100 Off | 00000000:2D:00.0 On | Off |
| 44% 59C P2 48W / 250W | 1885MiB / 32499MiB | 0% Default |
| | | N/A |
+-------------------------------+----------------------+----------------------+
+-----------------------------------------------------------------------------+
| Processes: |
| GPU GI CI PID Type Process name GPU Memory |
| ID ID Usage |
|=============================================================================|
+-----------------------------------------------------------------------------+
[25]:
CATEGORICAL_COLUMNS = [
'hashtags', 'tweet_id', 'media', 'links', 'domains', 'tweet_type', 'language',
'a_user_id', 'a_is_verified',
'b_user_id', 'b_is_verified',
'b_follows_a', 'timestamp_wd'
]
CONTINUOUS_COLUMNS = [
'timestamp',
'a_follower_count', 'a_following_count',
'b_follower_count', 'b_following_count',
'hashtags_count_t', 'domains_count_t', 'links_count_t'
]
LABEL_COLUMNS = [
'reply', 'retweet', 'retweet_comment', 'like'
]
We initalize the NVTabular workflow.
[26]:
proc = nvt.Workflow(
cat_names=CATEGORICAL_COLUMNS,
cont_names=CONTINUOUS_COLUMNS,
label_name=LABEL_COLUMNS
)
We apply TargetEncoding with kfold of 5 and smoothing of 20. TargetEncoding is explained in here and here. We need to transform the LABEL_COLUMNS into boolean (0/1) targets.
[27]:
proc.add_feature(
[nvt.ops.LambdaOp(
op_name = 'change',
f = lambda col, gdf: (col>0).astype('int8'),
columns = LABEL_COLUMNS,
replace=True
),
nvt.ops.TargetEncoding(
cat_groups = ['media',
'tweet_type',
'language',
'a_user_id',
'b_user_id',
['domains','language','b_follows_a','tweet_type','media','a_is_verified']],
cont_target = LABEL_COLUMNS,
kfold = 5,
p_smooth = 20
)]
)
We count encode the columns media, tweet_type, language, a_user_id, b_user_id. CountEncoding is explained here.
[28]:
proc.add_preprocess(
nvt.ops.JoinGroupby(
columns = [
'media',
'tweet_type',
'language',
'a_user_id',
'b_user_id'
]
)
)
We transform timestamp to datetime type.
[29]:
proc.add_preprocess(
nvt.ops.LambdaOp(
op_name = 'to_datetime',
f = lambda col, gdf: cudf.to_datetime(col.astype('int32'), unit='s'),
columns = ['timestamp'],
replace=False
)
)
We extract hour from datetime. We extract minute from datetime. We extract seconds from datetime
[30]:
proc.add_preprocess(
nvt.ops.LambdaOp(
op_name = 'to_hour',
f = lambda col, gdf: col.dt.hour,
columns = ['timestamp_to_datetime'],
replace=False
)
)
proc.add_preprocess(
nvt.ops.LambdaOp(
op_name = 'to_minute',
f = lambda col, gdf: col.dt.minute,
columns = ['timestamp_to_datetime'],
replace=False
)
)
proc.add_preprocess(
nvt.ops.LambdaOp(
op_name = 'to_second',
f = lambda col, gdf: col.dt.second,
columns = ['timestamp_to_datetime'],
replace=False
)
)
We difference encode b_follower_count, b_following_count, language grouped by b_user_id. DifferenceEncoding is explained here. First, we need to transform the datatype to float32 to prevent overflow/underflow. After DifferenceEncoding, we want to fill NaN values with 0.
[31]:
proc.add_preprocess([
nvt.ops.LambdaOp(
op_name = 'asfloat',
f = lambda col, gdf: col.astype('float32'),
columns = ['b_follower_count','b_following_count','language'],
replace=True
),
nvt.ops.DifferenceLag(
'b_user_id',
columns=['b_follower_count','b_following_count','language'],
shift = [1, -1]
),
nvt.ops.FillMissing(fill_val=0)
])
We initialize the train and valid as NVTabular datasets.
[32]:
train_dataset = nvt.Dataset(glob.glob(BASE_DIR + 'nv_train/*.parquet'),
engine='parquet',
part_size="2GB")
valid_dataset = nvt.Dataset(glob.glob(BASE_DIR + 'nv_valid/*.parquet'),
engine='parquet',
part_size="2GB")
The columns a_is_verified, b_is_verified and b_follows_a have the datatype boolean. XGBoost does not support boolean datatypes and we need convert them to int8. We can define the output datatypes as a dict and parse it to the apply function.
[33]:
dict_dtypes = {}
for col in ['a_is_verified','b_is_verified','b_follows_a']:
dict_dtypes[col] = np.int8
We apply the workflow to the datasets.
[34]:
%%time
time_fe_start = time.time()
proc.apply(train_dataset, record_stats=True, output_path=BASE_DIR + 'nv_train_fe/', dtypes=dict_dtypes)
proc.apply(valid_dataset, record_stats=False, output_path=BASE_DIR + 'nv_valid_fe/', dtypes=dict_dtypes)
time_fe = time.time()-time_fe_start
/opt/conda/envs/rapids/lib/python3.7/site-packages/cudf/core/join/join.py:368: UserWarning: can't safely cast column from right with type int64 to uint32, upcasting to int64
"right", dtype_r, dtype_l, libcudf_join_type
/opt/conda/envs/rapids/lib/python3.7/site-packages/dask/dataframe/io/parquet/core.py:414: UserWarning: If read back by Dask, column named __null_dask_index__ will be set to the index (and renamed to None).
"If read back by Dask, column named __null_dask_index__ "
CPU times: user 2min 27s, sys: 1min 20s, total: 3min 48s
Wall time: 3min 54s
Training our model
After the preprocessing and feature engineering is done, we can train a model to predict our targets. We load our datasets with dask_cudf.
[35]:
train = dask_cudf.read_parquet(BASE_DIR + 'nv_train_fe/*.parquet')
valid = dask_cudf.read_parquet(BASE_DIR + 'nv_valid_fe/*.parquet')
[36]:
train[['a_is_verified','b_is_verified','b_follows_a']].dtypes
[36]:
a_is_verified int8
b_is_verified int8
b_follows_a int8
dtype: object
Some columns are only used for feature engineering. Therefore, we define the columns we want to ignore for training.
[37]:
dont_use =[
'__null_dask_index__',
'text_tokens',
'timestamp',
'a_account_creation',
'b_account_creation',
'hashtags',
'tweet_id',
'links',
'domains',
'a_user_id',
'b_user_id',
'timestamp_wd',
'timestamp_to_datetime',
'a_following_count_a_ff_rate',
'b_following_count_b_ff_rate'
]
dont_use = [x for x in train.columns if x in dont_use]
label_names = ['reply', 'retweet', 'retweet_comment', 'like']
We drop the columns, which are not required for training.
Our experiments show that we require only 10% of the training dataset. Our feature engineering, such as TargetEncoding, uses the training datasets and leverage the information of the full dataset. In the competition, we trained our models with higher ratio (20% and 50%), but we could not observe an improvement in performance. We sample the training dataset to 10% of the size and drop all columns, which we do not want to use.
[38]:
SAMPLE_RATIO = 0.1
SEED = 1
if SAMPLE_RATIO < 1.0:
train['sample'] = train['tweet_id'].map_partitions(lambda cudf_df: cudf_df.hash_encode(stop=10))
print(len(train))
train = train[train['sample']<10*SAMPLE_RATIO]
train, = dask.persist(train)
print(len(train))
Y_train = train[label_names]
Y_train, = dask.persist(Y_train)
train = train.drop(['sample']+label_names+dont_use,axis=1)
train, = dask.persist(train)
print('Using %i features'%(train.shape[1]))
67513999
6771556
Using 51 features
Similar to the training dataset, our experiments show that 35% of our validation dataset is enough to get a good estimate of the performance metric. 35% of the validation dataset has a similar size as the test set of the RecSys2020 competition.
[39]:
SAMPLE_RATIO = 0.35
SEED = 1
if SAMPLE_RATIO < 1.0:
print(len(valid))
valid['sample'] = valid['tweet_id'].map_partitions(lambda cudf_df: cudf_df.hash_encode(stop=10))
valid = valid[valid['sample']<10*SAMPLE_RATIO]
valid, = dask.persist(valid)
print(len(valid))
Y_valid = valid[label_names]
Y_valid, = dask.persist(Y_valid)
valid = valid.drop(['sample']+label_names+dont_use,axis=1)
valid, = dask.persist(valid)
25814765
10320615
We initialize our XGBoost parameter.
[40]:
print('XGB Version',xgb.__version__)
xgb_parms = {
'max_depth':8,
'learning_rate':0.1,
'subsample':0.8,
'colsample_bytree':0.3,
'eval_metric':'logloss',
'objective':'binary:logistic',
'tree_method':'gpu_hist',
'predictor' : 'gpu_predictor'
}
XGB Version 1.2.0-SNAPSHOT
[41]:
train,valid = dask.persist(train,valid)
We train our XGBoost models. The challenge requires to predict 4 targets, does a user 1. like a tweet 2. reply a tweet 3. comment a tweet 4. comment and reply a tweet
We train 4x XGBoost models for 300 rounds on a GPU.
[42]:
%%time
time_train_start = time.time()
NROUND = 300
VERBOSE_EVAL = 50
preds = []
for i in range(4):
name = label_names[i]
print('#'*25);print('###',name);print('#'*25)
start = time.time(); print('Creating DMatrix...')
dtrain = xgb.dask.DaskDMatrix(client,data=train,label=Y_train.iloc[:, i])
print('Took %.1f seconds'%(time.time()-start))
start = time.time(); print('Training...')
model = xgb.dask.train(client, xgb_parms,
dtrain=dtrain,
num_boost_round=NROUND,
verbose_eval=VERBOSE_EVAL)
print('Took %.1f seconds'%(time.time()-start))
start = time.time(); print('Predicting...')
preds.append(xgb.dask.predict(client,model,valid))
print('Took %.1f seconds'%(time.time()-start))
del model, dtrain
time_train = time.time()-time_train_start
#########################
### reply
#########################
Creating DMatrix...
Took 0.8 seconds
Training...
/opt/conda/envs/rapids/lib/python3.7/site-packages/distributed/client.py:3518: RuntimeWarning: coroutine 'Client._update_scheduler_info' was never awaited
self.sync(self._update_scheduler_info)
Took 13.6 seconds
Predicting...
/opt/conda/envs/rapids/lib/python3.7/site-packages/distributed/worker.py:3385: UserWarning: Large object of size 4.13 MB detected in task graph:
[<function _predict_async.<locals>.mapped_predict ... titions>, True]
Consider scattering large objects ahead of time
with client.scatter to reduce scheduler burden and
keep data on workers
future = client.submit(func, big_data) # bad
big_future = client.scatter(big_data) # good
future = client.submit(func, big_future) # good
% (format_bytes(len(b)), s)
Took 0.4 seconds
#########################
### retweet
#########################
Creating DMatrix...
Took 0.1 seconds
Training...
/opt/conda/envs/rapids/lib/python3.7/site-packages/distributed/client.py:3518: RuntimeWarning: coroutine 'Client._update_scheduler_info' was never awaited
self.sync(self._update_scheduler_info)
Took 13.6 seconds
Predicting...
Took 0.4 seconds
#########################
### retweet_comment
#########################
Creating DMatrix...
Took 0.1 seconds
Training...
/opt/conda/envs/rapids/lib/python3.7/site-packages/distributed/client.py:3518: RuntimeWarning: coroutine 'Client._update_scheduler_info' was never awaited
self.sync(self._update_scheduler_info)
Took 12.5 seconds
Predicting...
Took 0.3 seconds
#########################
### like
#########################
Creating DMatrix...
Took 0.1 seconds
Training...
/opt/conda/envs/rapids/lib/python3.7/site-packages/distributed/client.py:3518: RuntimeWarning: coroutine 'Client._update_scheduler_info' was never awaited
self.sync(self._update_scheduler_info)
Took 13.7 seconds
Predicting...
Took 0.5 seconds
CPU times: user 3.89 s, sys: 40.2 ms, total: 3.93 s
Wall time: 56.3 s
[43]:
yvalid = Y_valid[label_names].values.compute()
oof = cp.array([i.values.compute() for i in preds]).T
[44]:
yvalid.shape
[44]:
(10320615, 4)
The hosts of the RecSys2020 competition provide code for calculating the performance metric PRAUC and RCE. We optimized the code to speed up the calculation, as well. Using cuDF / cupy, we calculate the performance metric on the GPU.
[45]:
from sklearn.metrics import auc
def precision_recall_curve(y_true,y_pred):
y_true = y_true.astype('float32')
ids = cp.argsort(-y_pred)
y_true = y_true[ids]
y_pred = y_pred[ids]
y_pred = cp.flip(y_pred,axis=0)
acc_one = cp.cumsum(y_true)
sum_one = cp.sum(y_true)
precision = cp.flip(acc_one/cp.cumsum(cp.ones(len(y_true))),axis=0)
precision[:-1] = precision[1:]
precision[-1] = 1.
recall = cp.flip(acc_one/sum_one,axis=0)
recall[:-1] = recall[1:]
recall[-1] = 0
n = (recall==1).sum()
return precision[n-1:],recall[n-1:],y_pred[n:]
def compute_prauc(pred, gt):
prec, recall, thresh = precision_recall_curve(gt, pred)
recall, prec = cp.asnumpy(recall), cp.asnumpy(prec)
prauc = auc(recall, prec)
return prauc
def log_loss(y_true,y_pred,eps=1e-7, normalize=True, sample_weight=None):
y_true = y_true.astype('int32')
y_pred = cp.clip(y_pred, eps, 1 - eps)
if y_pred.ndim == 1:
y_pred = cp.expand_dims(y_pred, axis=1)
if y_pred.shape[1] == 1:
y_pred = cp.hstack([1 - y_pred, y_pred])
y_pred /= cp.sum(y_pred, axis=1, keepdims=True)
loss = -cp.log(y_pred)[cp.arange(y_pred.shape[0]), y_true]
return _weighted_sum(loss, sample_weight, normalize).item()
def _weighted_sum(sample_score, sample_weight, normalize):
if normalize:
return cp.average(sample_score, weights=sample_weight)
elif sample_weight is not None:
return cp.dot(sample_score, sample_weight)
else:
return sample_score.sum()
def compute_rce_fast(pred, gt):
cross_entropy = log_loss(gt, pred)
yt = cp.mean(gt).item()
# cross_entropy and yt are single numbers (no arrays) and using CPU is fast
strawman_cross_entropy = -(yt*np.log(yt) + (1 - yt)*np.log(1 - yt))
return (1.0 - cross_entropy/strawman_cross_entropy)*100.0
Finally, we calculate the performance metric PRAUC and RCE for each target.
[46]:
txt = ''
for i in range(4):
prauc = compute_prauc(oof[:,i], yvalid[:, i])
rce = compute_rce_fast(oof[:,i], yvalid[:, i]).item()
txt_ = f"{label_names[i]:20} PRAUC:{prauc:.5f} RCE:{rce:.5f}"
print(txt_)
txt += txt_ + '\n'
reply PRAUC:0.13551 RCE:16.77993
retweet PRAUC:0.51663 RCE:27.94596
retweet_comment PRAUC:0.05005 RCE:10.55952
like PRAUC:0.76354 RCE:22.98230
[47]:
time_total = time.time()-time_total_start
[48]:
print('Total time: {:.2f}s'.format(time_total))
print()
print('1. Preprocessing: {:.2f}s'.format(time_preproc))
print('2. Splitting: {:.2f}s'.format(time_split))
print('3. Feature engineering: {:.2f}s'.format(time_fe))
print('4. Training: {:.2f}s'.format(time_train))
Total time: 780.36s
1. Preprocessing: 337.52s
2. Splitting: 122.69s
3. Feature engineering: 234.35s
4. Training: 56.31s