"test/pipelines/pipelines-it-remote.yml" did not exist on "ae0d36faad275281995a565ff470aa33aef922d8"
ncf_main.py 16.5 KB
Newer Older
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
# Copyright 2018 The TensorFlow Authors. All Rights Reserved.
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
#     http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
# ==============================================================================
"""NCF framework to train and evaluate the NeuMF model.

The NeuMF model assembles both MF and MLP models under the NCF framework. Check
`neumf_model.py` for more details about the models.
"""
20

21
22
23
24
from __future__ import absolute_import
from __future__ import division
from __future__ import print_function

25
import contextlib
26
import heapq
27
28
import json
import logging
29
import math
30
import multiprocessing
31
import os
32
33
import signal
import typing
34

35
# pylint: disable=g-bad-import-order
36
import numpy as np
37
38
from absl import app as absl_app
from absl import flags
39
import tensorflow as tf
40
# pylint: enable=g-bad-import-order
41

Reed's avatar
Reed committed
42
from tensorflow.contrib.compiler import xla
43
from official.datasets import movielens
44
from official.recommendation import constants as rconst
45
from official.recommendation import data_pipeline
46
from official.recommendation import data_preprocessing
47
from official.recommendation import neumf_model
48
49
50
from official.utils.flags import core as flags_core
from official.utils.logs import hooks_helper
from official.utils.logs import logger
51
from official.utils.logs import mlperf_helper
52
from official.utils.misc import distribution_utils
53
from official.utils.misc import model_helpers
54
55


Reed's avatar
Reed committed
56
57
58
FLAGS = flags.FLAGS


59
def construct_estimator(model_dir, params):
60
  """Construct either an Estimator or TPUEstimator for NCF.
61
62

  Args:
63
64
    model_dir: The model directory for the estimator
    params: The params dict for the estimator
65
66

  Returns:
67
    An Estimator or TPUEstimator.
68
69
  """

70
  if params["use_tpu"]:
71
72
73
74
75
    # Some of the networking libraries are quite chatty.
    for name in ["googleapiclient.discovery", "googleapiclient.discovery_cache",
                 "oauth2client.transport"]:
      logging.getLogger(name).setLevel(logging.ERROR)

76
77
78
79
    tpu_cluster_resolver = tf.contrib.cluster_resolver.TPUClusterResolver(
        tpu=params["tpu"],
        zone=params["tpu_zone"],
        project=params["tpu_gcp_project"],
80
        coordinator_name="coordinator"
81
    )
82

83
84
    tf.logging.info("Issuing reset command to TPU to ensure a clean state.")
    tf.Session.reset(tpu_cluster_resolver.get_master())
85

86
87
88
89
90
91
92
93
94
95
96
97
    # Estimator looks at the master it connects to for MonitoredTrainingSession
    # by reading the `TF_CONFIG` environment variable, and the coordinator
    # is used by StreamingFilesDataset.
    tf_config_env = {
        "session_master": tpu_cluster_resolver.get_master(),
        "eval_session_master": tpu_cluster_resolver.get_master(),
        "coordinator": tpu_cluster_resolver.cluster_spec()
                       .as_dict()["coordinator"]
    }
    os.environ['TF_CONFIG'] = json.dumps(tf_config_env)

    distribution = tf.contrib.distribute.TPUStrategy(
Taylor Robie's avatar
Taylor Robie committed
98
        tpu_cluster_resolver, steps_per_run=100)
99
100
101
102
103

  else:
    distribution = distribution_utils.get_distribution_strategy(
        num_gpus=params["num_gpus"])

104
105
  run_config = tf.estimator.RunConfig(train_distribute=distribution,
                                      eval_distribute=distribution)
106

Reed's avatar
Reed committed
107
108
109
110
111
112
  model_fn = neumf_model.neumf_model_fn
  if params["use_xla_for_gpu"]:
    tf.logging.info("Using XLA for GPU for training and evaluation.")
    model_fn = xla.estimator_model_fn(model_fn)
  estimator = tf.estimator.Estimator(model_fn=model_fn, model_dir=model_dir,
                                     config=run_config, params=params)
113
114
115
116
  return estimator


def log_and_get_hooks(eval_batch_size):
Taylor Robie's avatar
Taylor Robie committed
117
  """Convenience function for hook and logger creation."""
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
  # Create hooks that log information about the training and metric values
  train_hooks = hooks_helper.get_train_hooks(
      FLAGS.hooks,
      model_dir=FLAGS.model_dir,
      batch_size=FLAGS.batch_size,  # for ExamplesPerSecondHook
      tensors_to_log={"cross_entropy": "cross_entropy"}
  )
  run_params = {
      "batch_size": FLAGS.batch_size,
      "eval_batch_size": eval_batch_size,
      "number_factors": FLAGS.num_factors,
      "hr_threshold": FLAGS.hr_threshold,
      "train_epochs": FLAGS.train_epochs,
  }
  benchmark_logger = logger.get_benchmark_logger()
  benchmark_logger.log_run_info(
      model_name="recommendation",
      dataset_name=FLAGS.dataset,
      run_params=run_params,
      test_id=FLAGS.benchmark_test_id)

  return benchmark_logger, train_hooks


def parse_flags(flags_obj):
Taylor Robie's avatar
Taylor Robie committed
143
  """Convenience function to turn flags into params."""
144
145
146
  num_gpus = flags_core.get_num_gpus(flags_obj)
  num_devices = FLAGS.num_tpu_shards if FLAGS.tpu else num_gpus or 1

Taylor Robie's avatar
Taylor Robie committed
147
  batch_size = (flags_obj.batch_size + num_devices - 1) // num_devices
148
149

  eval_divisor = (rconst.NUM_EVAL_NEGATIVES + 1) * num_devices
Taylor Robie's avatar
Taylor Robie committed
150
151
152
  eval_batch_size = flags_obj.eval_batch_size or flags_obj.batch_size
  eval_batch_size = ((eval_batch_size + eval_divisor - 1) //
                     eval_divisor * eval_divisor // num_devices)
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177

  return {
      "train_epochs": flags_obj.train_epochs,
      "batches_per_step": num_devices,
      "use_seed": flags_obj.seed is not None,
      "batch_size": batch_size,
      "eval_batch_size": eval_batch_size,
      "learning_rate": flags_obj.learning_rate,
      "mf_dim": flags_obj.num_factors,
      "model_layers": [int(layer) for layer in flags_obj.layers],
      "mf_regularization": flags_obj.mf_regularization,
      "mlp_reg_layers": [float(reg) for reg in flags_obj.mlp_regularization],
      "num_neg": flags_obj.num_neg,
      "num_gpus": num_gpus,
      "use_tpu": flags_obj.tpu is not None,
      "tpu": flags_obj.tpu,
      "tpu_zone": flags_obj.tpu_zone,
      "tpu_gcp_project": flags_obj.tpu_gcp_project,
      "beta1": flags_obj.beta1,
      "beta2": flags_obj.beta2,
      "epsilon": flags_obj.epsilon,
      "match_mlperf": flags_obj.ml_perf,
      "use_xla_for_gpu": flags_obj.use_xla_for_gpu,
      "epochs_between_evals": FLAGS.epochs_between_evals,
  }
178
179
180


def main(_):
Reed's avatar
Reed committed
181
182
  with logger.benchmark_context(FLAGS), \
       mlperf_helper.LOGGER(FLAGS.output_ml_perf_compliance_logging):
183
    mlperf_helper.set_ncf_root(os.path.split(os.path.abspath(__file__))[0])
184
185
186
187
188
    run_ncf(FLAGS)


def run_ncf(_):
  """Run NCF training and eval loop."""
189
  if FLAGS.download_if_missing and not FLAGS.use_synthetic_data:
190
    movielens.download(FLAGS.dataset, FLAGS.data_dir)
191

192
193
194
  if FLAGS.seed is not None:
    np.random.seed(FLAGS.seed)

195
  params = parse_flags(FLAGS)
196
  total_training_cycle = FLAGS.train_epochs // FLAGS.epochs_between_evals
197

198
  if FLAGS.use_synthetic_data:
199
    producer = data_pipeline.DummyConstructor()
200
201
    num_users, num_items = data_preprocessing.DATASET_TO_NUM_USERS_AND_ITEMS[
        FLAGS.dataset]
202
203
    num_train_steps = rconst.SYNTHETIC_BATCHES_PER_EPOCH
    num_eval_steps = rconst.SYNTHETIC_BATCHES_PER_EPOCH
204
  else:
205
    num_users, num_items, producer = data_preprocessing.instantiate_pipeline(
206
207
208
        dataset=FLAGS.dataset, data_dir=FLAGS.data_dir, params=params,
        constructor_type=FLAGS.constructor_type,
        deterministic=FLAGS.seed is not None)
209

210
211
212
213
214
215
216
    num_train_steps = (producer.train_batches_per_epoch //
                       params["batches_per_step"])
    num_eval_steps = (producer.eval_batches_per_epoch //
                      params["batches_per_step"])
    assert not producer.train_batches_per_epoch % params["batches_per_step"]
    assert not producer.eval_batches_per_epoch % params["batches_per_step"]
  producer.start()
217

218
219
  params["num_users"], params["num_items"] = num_users, num_items
  model_helpers.apply_clean(flags.FLAGS)
220

221
  estimator = construct_estimator(model_dir=FLAGS.model_dir, params=params)
222

223
  benchmark_logger, train_hooks = log_and_get_hooks(params["eval_batch_size"])
224

225
226
  target_reached = False
  mlperf_helper.ncf_print(key=mlperf_helper.TAGS.TRAIN_LOOP)
227
  for cycle_index in range(total_training_cycle):
228
    assert FLAGS.epochs_between_evals == 1 or not mlperf_helper.LOGGER.enabled
229
    tf.logging.info("Starting a training cycle: {}/{}".format(
230
        cycle_index + 1, total_training_cycle))
231

232
233
234
    mlperf_helper.ncf_print(key=mlperf_helper.TAGS.TRAIN_EPOCH,
                            value=cycle_index)

235
236
237
238
239
240
241
242
243
244
245
246
    train_input_fn = producer.make_input_fn(is_training=True)
    estimator.train(input_fn=train_input_fn, hooks=train_hooks,
                    steps=num_train_steps)

    tf.logging.info("Beginning evaluation.")
    eval_input_fn = producer.make_input_fn(is_training=False)

    mlperf_helper.ncf_print(key=mlperf_helper.TAGS.EVAL_START,
                            value=cycle_index)
    eval_results = estimator.evaluate(eval_input_fn, steps=num_eval_steps)
    tf.logging.info("Evaluation complete.")

247
248
    hr = float(eval_results[rconst.HR_KEY])
    ndcg = float(eval_results[rconst.NDCG_KEY])
249
    loss = float(eval_results["loss"])
250

251
252
253
254
255
256
257
258
259
260
261
    mlperf_helper.ncf_print(
        key=mlperf_helper.TAGS.EVAL_TARGET,
        value={"epoch": cycle_index, "value": FLAGS.hr_threshold})
    mlperf_helper.ncf_print(key=mlperf_helper.TAGS.EVAL_ACCURACY,
                            value={"epoch": cycle_index, "value": hr})
    mlperf_helper.ncf_print(
        key=mlperf_helper.TAGS.EVAL_HP_NUM_NEG,
        value={"epoch": cycle_index, "value": rconst.NUM_EVAL_NEGATIVES})

    mlperf_helper.ncf_print(key=mlperf_helper.TAGS.EVAL_STOP, value=cycle_index)

262
263
264
    # Benchmark the evaluation results
    benchmark_logger.log_evaluation_result(eval_results)
    # Log the HR and NDCG results.
265
    tf.logging.info(
266
267
        "Iteration {}: HR = {:.4f}, NDCG = {:.4f}, Loss = {:.4f}".format(
            cycle_index + 1, hr, ndcg, loss))
268
269
270

    # If some evaluation threshold is met
    if model_helpers.past_stop_threshold(FLAGS.hr_threshold, hr):
271
      target_reached = True
272
273
      break

274
275
  mlperf_helper.ncf_print(key=mlperf_helper.TAGS.RUN_STOP,
                          value={"success": target_reached})
276
277
  producer.stop_loop()
  producer.join()
278

279
280
281
  # Clear the session explicitly to avoid session delete error
  tf.keras.backend.clear_session()

282
283
  mlperf_helper.ncf_print(key=mlperf_helper.TAGS.RUN_FINAL)

284
285
286
287
288
289
290
291
292

def define_ncf_flags():
  """Add flags for running ncf_main."""
  # Add common flags
  flags_core.define_base(export_dir=False)
  flags_core.define_performance(
      num_parallel_calls=False,
      inter_op=False,
      intra_op=False,
293
      synthetic_data=True,
294
      max_train_steps=False,
295
296
      dtype=False,
      all_reduce_alg=False
297
  )
298
  flags_core.define_device(tpu=True)
299
300
301
302
303
304
305
306
307
  flags_core.define_benchmark()

  flags.adopt_module_key_flags(flags_core)

  flags_core.set_defaults(
      model_dir="/tmp/ncf/",
      data_dir="/tmp/movielens-data/",
      train_epochs=2,
      batch_size=256,
308
309
310
      hooks="ProfilerHook",
      tpu=None
  )
311
312
313
314
315
316
317
318

  # Add ncf-specific flags
  flags.DEFINE_enum(
      name="dataset", default="ml-1m",
      enum_values=["ml-1m", "ml-20m"], case_sensitive=False,
      help=flags_core.help_wrap(
          "Dataset to be trained and evaluated."))

319
320
321
322
  flags.DEFINE_boolean(
      name="download_if_missing", default=True, help=flags_core.help_wrap(
          "Download data to data_dir if it is not already present."))

323
  flags.DEFINE_integer(
324
325
326
327
328
329
      name="eval_batch_size", default=None, help=flags_core.help_wrap(
          "The batch size used for evaluation. This should generally be larger"
          "than the training batch size as the lack of back propagation during"
          "evaluation can allow for larger batch sizes to fit in memory. If not"
          "specified, the training batch size (--batch_size) will be used."))

330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
  flags.DEFINE_integer(
      name="num_factors", default=8,
      help=flags_core.help_wrap("The Embedding size of MF model."))

  # Set the default as a list of strings to be consistent with input arguments
  flags.DEFINE_list(
      name="layers", default=["64", "32", "16", "8"],
      help=flags_core.help_wrap(
          "The sizes of hidden layers for MLP. Example "
          "to specify different sizes of MLP layers: --layers=32,16,8,4"))

  flags.DEFINE_float(
      name="mf_regularization", default=0.,
      help=flags_core.help_wrap(
          "The regularization factor for MF embeddings. The factor is used by "
          "regularizer which allows to apply penalties on layer parameters or "
          "layer activity during optimization."))

  flags.DEFINE_list(
      name="mlp_regularization", default=["0.", "0.", "0.", "0."],
      help=flags_core.help_wrap(
          "The regularization factor for each MLP layer. See mf_regularization "
          "help for more info about regularization factor."))

  flags.DEFINE_integer(
      name="num_neg", default=4,
      help=flags_core.help_wrap(
          "The Number of negative instances to pair with a positive instance."))

  flags.DEFINE_float(
      name="learning_rate", default=0.001,
      help=flags_core.help_wrap("The learning rate."))

363
364
365
366
367
368
369
370
371
372
373
374
375
  flags.DEFINE_float(
      name="beta1", default=0.9,
      help=flags_core.help_wrap("beta1 hyperparameter for the Adam optimizer."))

  flags.DEFINE_float(
      name="beta2", default=0.999,
      help=flags_core.help_wrap("beta2 hyperparameter for the Adam optimizer."))

  flags.DEFINE_float(
      name="epsilon", default=1e-8,
      help=flags_core.help_wrap("epsilon hyperparameter for the Adam "
                                "optimizer."))

376
377
378
379
380
381
382
383
  flags.DEFINE_float(
      name="hr_threshold", default=None,
      help=flags_core.help_wrap(
          "If passed, training will stop when the evaluation metric HR is "
          "greater than or equal to hr_threshold. For dataset ml-1m, the "
          "desired hr_threshold is 0.68 which is the result from the paper; "
          "For dataset ml-20m, the threshold can be set as 0.95 which is "
          "achieved by MLPerf implementation."))
384

385
386
387
388
389
390
391
392
  flags.DEFINE_enum(
      name="constructor_type", default="bisection",
      enum_values=["bisection", "materialized"], case_sensitive=False,
      help=flags_core.help_wrap(
          "Strategy to use for generating false negatives. materialized has a"
          "precompute that scales badly, but a faster per-epoch construction"
          "time and can be faster on very large systems."))

393
  flags.DEFINE_bool(
394
      name="ml_perf", default=False,
395
396
397
398
399
400
401
402
403
404
405
406
407
      help=flags_core.help_wrap(
          "If set, changes the behavior of the model slightly to match the "
          "MLPerf reference implementations here: \n"
          "https://github.com/mlperf/reference/tree/master/recommendation/"
          "pytorch\n"
          "The two changes are:\n"
          "1. When computing the HR and NDCG during evaluation, remove "
          "duplicate user-item pairs before the computation. This results in "
          "better HRs and NDCGs.\n"
          "2. Use a different soring algorithm when sorting the input data, "
          "which performs better due to the fact the sorting algorithms are "
          "not stable."))

Reed's avatar
Reed committed
408
409
410
411
412
413
414
415
416
417
418
419
  flags.DEFINE_bool(
      name="output_ml_perf_compliance_logging", default=False,
      help=flags_core.help_wrap(
          "If set, output the MLPerf compliance logging. This is only useful "
          "if one is running the model for MLPerf. See "
          "https://github.com/mlperf/policies/blob/master/training_rules.adoc"
          "#submission-compliance-logs for details. This uses sudo and so may "
          "ask for your password, as root access is needed to clear the system "
          "caches, which is required for MLPerf compliance."
      )
  )

420
421
422
423
  flags.DEFINE_integer(
      name="seed", default=None, help=flags_core.help_wrap(
          "This value will be used to seed both NumPy and TensorFlow."))

424
425
426
  @flags.validator("eval_batch_size", "eval_batch_size must be at least {}"
                   .format(rconst.NUM_EVAL_NEGATIVES + 1))
  def eval_size_check(eval_batch_size):
Taylor Robie's avatar
Taylor Robie committed
427
428
    return (eval_batch_size is None or
            int(eval_batch_size) > rconst.NUM_EVAL_NEGATIVES)
429

Reed's avatar
Reed committed
430
431
432
433
434
  flags.DEFINE_bool(
      name="use_xla_for_gpu", default=False, help=flags_core.help_wrap(
          "If True, use XLA for the model function. Only works when using a "
          "GPU. On TPUs, XLA is always used"))

435
436
437
438
  xla_message = "--use_xla_for_gpu is incompatible with --tpu"
  @flags.multi_flags_validator(["use_xla_for_gpu", "tpu"], message=xla_message)
  def xla_validator(flag_dict):
    return not flag_dict["use_xla_for_gpu"] or not flag_dict["tpu"]
Reed's avatar
Reed committed
439

440
441
442

if __name__ == "__main__":
  tf.logging.set_verbosity(tf.logging.INFO)
443
444
  define_ncf_flags()
  absl_app.run(main)