DCausse has submitted this change and it was merged. (
https://gerrit.wikimedia.org/r/364637 )
Change subject: Calculate CV splits on the driver
......................................................................
Calculate CV splits on the driver
The calculation of CV splits is only deterministic if it is
called each time with the exact same rows per partition. This
guarnatee is not met because catalyst decides how many partitions
a parquet df read from disk should have based on the query plan. This
led to intermittent information leakage between test and train, and
ocasionally the groupData passed to xgboost being incorrect if the
splits were re-calculated after calculating the groupData.
This new solution collects all the rows to the driver and calculates
the splits there. This guarantees the splits are calculated a single
time and any re-calculation will use the provided splits.
Change-Id: I8a8189f999d5a59cbf699b39b98b4e3a4ce0549b
---
M mjolnir/cli/training_pipeline.py
A mjolnir/feature_engineering.py
M mjolnir/test/training/test_hyperopt.py
M mjolnir/test/training/test_tuning.py
M mjolnir/training/hyperopt.py
M mjolnir/training/tuning.py
M mjolnir/training/xgboost.py
7 files changed, 42 insertions(+), 49 deletions(-)
Approvals:
DCausse: Verified; Looks good to me, approved
diff --git a/mjolnir/cli/training_pipeline.py b/mjolnir/cli/training_pipeline.py
index dab8301..2304de2 100644
--- a/mjolnir/cli/training_pipeline.py
+++ b/mjolnir/cli/training_pipeline.py
@@ -34,15 +34,10 @@
print ''
continue
- # Make a guess at the number of fold partitions needed based on data
size.
- # This requires there to be around 40k data points, arbitrarily chosen,
- # per partition used to calculate the folds, up to 100 partitions.
- num_fold_partitions = min(100, max(1, data_size / 40000))
-
# Explore a hyperparameter space. Skip the most expensive part of
tuning,
# increasing the # of trees, with target_node_evaluations=None
tune_results = mjolnir.training.xgboost.tune(
- df_hits_with_features, num_folds=num_folds,
num_fold_partitions=num_fold_partitions,
+ df_hits_with_features, num_folds=num_folds,
num_cv_jobs=num_cv_jobs, num_workers=num_workers,
target_node_evaluations=target_node_evaluations)
diff --git a/mjolnir/feature_engineering.py b/mjolnir/feature_engineering.py
new file mode 100644
index 0000000..8b942d2
--- /dev/null
+++ b/mjolnir/feature_engineering.py
@@ -0,0 +1,12 @@
+import numpy as np
+from pyspark.ml.linalg import Vectors, VectorUDT
+
+def append_features(df, *cols):
+ def add_features(feat, *other):
+ raw = feat.toArray()
+ return Vectors.dense(np.append(raw, map(float, other)))
+ add_features_udf = F.udf(add_features, VectorUDT())
+ new_feat_list = df.schema['features'].metadata['features'] + cols
+ return df.withColumn('features', mjolnir.spark.add_meta(df._sc,
add_features_udf('features', *cols),
+ {'features':
new_feat_list}))
+
diff --git a/mjolnir/test/training/test_hyperopt.py
b/mjolnir/test/training/test_hyperopt.py
index f0bafad..ea2e334 100644
--- a/mjolnir/test/training/test_hyperopt.py
+++ b/mjolnir/test/training/test_hyperopt.py
@@ -31,7 +31,7 @@
# mock it out w/MockModel.
best_params, trails = mjolnir.training.hyperopt.minimize(
df_train, MockModel, space, max_evals=5, num_folds=2,
- num_fold_partitions=1, num_cv_jobs=1, num_workers=1)
+ num_cv_jobs=1, num_workers=1)
assert isinstance(best_params, dict)
# num_rounds should have been unchanged
assert 'num_rounds' in best_params
@@ -49,7 +49,7 @@
gen = MockModelGen()
best_params, trials = mjolnir.training.hyperopt.grid_search(
df_train, gen, space, num_folds=2,
- num_fold_partitions=1, num_cv_jobs=1, num_workers=1)
+ num_cv_jobs=1, num_workers=1)
assert isinstance(best_params, dict)
assert 'num_rounds' in best_params
# num rounds should be unchanged
diff --git a/mjolnir/test/training/test_tuning.py
b/mjolnir/test/training/test_tuning.py
index 73d9057..f0e8344 100644
--- a/mjolnir/test/training/test_tuning.py
+++ b/mjolnir/test/training/test_tuning.py
@@ -54,6 +54,6 @@
# xgboost needs all jobs to have a worker assigned before it will
# finish a round of training, so we have to be careful not to use
# too many workers
- num_folds=2, num_fold_partitions=1, num_cv_jobs=1, num_workers=1)
+ num_folds=2, num_cv_jobs=1, num_workers=1)
# one score for each fold
assert len(scores) == 2
diff --git a/mjolnir/training/hyperopt.py b/mjolnir/training/hyperopt.py
index 518497b..b15be07 100644
--- a/mjolnir/training/hyperopt.py
+++ b/mjolnir/training/hyperopt.py
@@ -37,18 +37,17 @@
return rval
-def grid_search(df, train_func, space, num_folds=5, num_fold_partitions=100,
- num_cv_jobs=5, num_workers=5):
+def grid_search(df, train_func, space, num_folds=5, num_cv_jobs=5,
num_workers=5):
# TODO: While this tried to look simple, hyperopt is a bit odd to integrate
# with this directly. Perhaps implement naive gridsearch directly instead
# of through hyperopt.
algo = _GridSearchAlgo(space)
return minimize(df, train_func, space, algo.max_evals, algo, num_folds,
- num_fold_partitions, num_cv_jobs, num_workers)
+ num_cv_jobs, num_workers)
def minimize(df, train_func, space, max_evals=50, algo=hyperopt.tpe.suggest,
- num_folds=5, num_fold_partitions=100, num_cv_jobs=5,
num_workers=5):
+ num_folds=5, num_cv_jobs=5, num_workers=5):
"""Perform cross validated hyperparameter optimization of train_func
Parameters
@@ -66,10 +65,6 @@
details.
num_folds : int
Number of folds to split df into for cross validation
- num_fold_partitions : int
- Sets the number of partitions to split with. Each partition needs
- to be some minimum size for averages to work out to an evenly split
- final set. (Default: 100)
num_cv_jobs : int
Number of cross validation folds to train in parallel
num_workers : int
@@ -112,8 +107,7 @@
}
folds = mjolnir.training.tuning._make_folds(
- df, num_folds=num_folds, num_workers=num_workers,
- num_fold_partitions=num_fold_partitions, num_cv_jobs=num_cv_jobs)
+ df, num_folds=num_folds, num_workers=num_workers,
num_cv_jobs=num_cv_jobs)
for fold in folds:
fold['train'].cache()
diff --git a/mjolnir/training/tuning.py b/mjolnir/training/tuning.py
index 7814748..8548f5d 100644
--- a/mjolnir/training/tuning.py
+++ b/mjolnir/training/tuning.py
@@ -48,7 +48,7 @@
mjolnir.spark.assert_columns(df, ['wikiid', 'norm_query_id'])
- def split_partition(rows):
+ def split_rows(rows):
# Current number of items per split
split_counts = defaultdict(lambda: [0] * len(splits))
# starting at 1 prevents div by zero. Using a float allows later
@@ -68,15 +68,22 @@
yield (row.wikiid, row.norm_query_id, 0)
processed[row.wikiid] += row.weight
- df_splits = (
+ # Calculating splits with mapPartitions is only deterministic if the # of
input
+ # partitions stays the same but it seems catalyst can sometimes decide to
change
+ # the number of partitions of the input if it comes from disk based on the
rest
+ # of the plan.
+ # This fights back by calculating everything on the driver. Maybe not ideal
+ # but seems to work as we can guarantee the splits are calculated a single
time.
+ # This is reasonable because even with 10's of M of samples there will only
+ # be a few hundred thousand (wikiid, norm_query_id) rows.
+ rows = (
df
.groupBy('wikiid', 'norm_query_id')
.agg(F.count(F.lit(1)).alias('weight'))
- # Could we guess the correct number of partitions instead? I'm not
- # sure though how it should be decided, and would require taking
- # an extra pass over the data.
- .coalesce(num_partitions)
- .rdd.mapPartitions(split_partition)
+ .collect())
+
+ df_splits = (
+ sc.parallelize(split_rows(rows))
.toDF(['wikiid', 'norm_query_id', output_column]))
return df.join(df_splits, how='inner', on=['wikiid', 'norm_query_id'])
@@ -106,7 +113,7 @@
})))
-def _make_folds(df, num_folds, num_fold_partitions, num_cv_jobs, num_workers):
+def _make_folds(df, num_folds, num_cv_jobs, num_workers):
"""Transform a DataFrame with assigned folds into many dataframes.
The results of split and group_k_fold emit a single dataframe with folds
@@ -128,10 +135,6 @@
num_folds : int
Number of folds to create. If a 'fold' column already exists in df
this will be ignored.
- num_fold_partitions : int
- Sets the number of partitions to split with. Each partition needs
- to be some minimum size for averages to work out to an evenly split
- final set.
num_cv_jobs: int
Number of folds to prepare in parallel.
num_workers : int
@@ -147,10 +150,10 @@
corresponding to group data needed by xgboost for train/eval.
"""
if 'fold' in df.columns:
- num_folds = df.schema['fold'].metadata['num_folds']
+ assert num_folds == df.schema['fold'].metadata['num_folds']
df_folds = df
else:
- df_folds = group_k_fold(df, num_folds, num_fold_partitions)
+ df_folds = group_k_fold(df, num_folds)
def job(fold):
condition = F.col('fold') == fold
@@ -229,8 +232,7 @@
return map(job_w_retry, folds)
-def cross_validate(df, train_func, params, num_folds=5,
num_fold_partitions=100,
- num_cv_jobs=5, num_workers=5):
+def cross_validate(df, train_func, params, num_folds=5, num_cv_jobs=5,
num_workers=5):
"""Perform cross-validation of the dataframe
Parameters
@@ -243,10 +245,6 @@
parameters to pass on to train_func
num_folds : int
Number of folds to split df into for cross validation
- num_fold_partitions : int, optional
- Sets the number of partitions to split with. Each partition needs
- to be some minimum size for averages to work out to an evenly split
- final set. (Default: 100)
num_cv_jobs : int
Number of cross validation folds to train in parallel
num_workers : int
@@ -259,6 +257,6 @@
correspond the the model evaluation metric for the train and test
data frames.
"""
- folds = _make_folds(df, num_folds, num_fold_partitions, num_cv_jobs,
num_workers)
+ folds = _make_folds(df, num_folds, num_cv_jobs, num_workers)
return _cross_validate(folds, train_func, params, num_cv_jobs=num_cv_jobs,
num_workers=num_workers)
diff --git a/mjolnir/training/xgboost.py b/mjolnir/training/xgboost.py
index f64b115..46b4c9b 100644
--- a/mjolnir/training/xgboost.py
+++ b/mjolnir/training/xgboost.py
@@ -391,8 +391,7 @@
return eta_pred[idx]
-def tune(df, num_folds=5, num_fold_partitions=100, num_cv_jobs=5,
num_workers=5,
- target_node_evaluations=5000):
+def tune(df, num_folds=5, num_cv_jobs=5, num_workers=5,
target_node_evaluations=5000):
"""Find appropriate hyperparameters for training df
This is far from perfect, hyperparameter tuning is a bit of a black art
@@ -414,10 +413,6 @@
df : pyspark.sql.DataFrame
num_folds : int, optional
The number of cross validation folds to use while tuning. (Default: 5)
- num_fold_partitions : int, optional
- The number of partitions to use when calculating folds. For small
- datasets this needs to be a reasonably small number. For medium to
- large the default is reasonable. (Default: 100)
num_cv_jobs : int, optional
The number of cross validation folds to train in parallel. (Default: 5)
num_workers : int, optional
@@ -444,8 +439,7 @@
def eval_space(space, max_evals):
"""Eval a space using standard hyperopt"""
best, trials = mjolnir.training.hyperopt.minimize(
- df, train, space, max_evals=max_evals,
- num_folds=num_folds, num_fold_partitions=num_fold_partitions,
+ df, train, space, max_evals=max_evals, num_folds=num_folds,
num_cv_jobs=num_cv_jobs, num_workers=num_workers)
for k, v in space.items():
if not np.isscalar(v):
@@ -455,7 +449,7 @@
def eval_space_grid(space):
"""Eval all points in the space via a grid search"""
best, trials = mjolnir.training.hyperopt.grid_search(
- df, train, space, num_folds=num_folds,
num_fold_partitions=num_fold_partitions,
+ df, train, space, num_folds=num_folds,
num_cv_jobs=num_cv_jobs, num_workers=num_workers)
for k, v in space.items():
if not np.isscalar(v):
--
To view, visit https://gerrit.wikimedia.org/r/364637
To unsubscribe, visit https://gerrit.wikimedia.org/r/settings
Gerrit-MessageType: merged
Gerrit-Change-Id: I8a8189f999d5a59cbf699b39b98b4e3a4ce0549b
Gerrit-PatchSet: 1
Gerrit-Project: search/MjoLniR
Gerrit-Branch: master
Gerrit-Owner: EBernhardson <[email protected]>
Gerrit-Reviewer: DCausse <[email protected]>
Gerrit-Reviewer: EBernhardson <[email protected]>
_______________________________________________
MediaWiki-commits mailing list
[email protected]
https://lists.wikimedia.org/mailman/listinfo/mediawiki-commits
