data_test.py 12.5 KB
Newer Older
Frederick Liu's avatar
Frederick Liu committed
1
# Copyright 2021 The TensorFlow Authors. All Rights Reserved.
2
3
4
5
6
7
8
9
10
11
12
13
#
# 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.
Frederick Liu's avatar
Frederick Liu committed
14

15
16
17
18
19
20
"""Test NCF data pipeline."""

from __future__ import absolute_import
from __future__ import division
from __future__ import print_function

21
from collections import defaultdict
22
import hashlib
23
24
import os

25
import mock
Hongkun Yu's avatar
Hongkun Yu committed
26

27
import numpy as np
28
import scipy.stats
29
30
31
32
import tensorflow as tf

from official.recommendation import constants as rconst
from official.recommendation import data_preprocessing
33
from official.recommendation import movielens
Hongkun Yu's avatar
Hongkun Yu committed
34
from official.recommendation import popen_helper
35
36
37
38
39
40

DATASET = "ml-test"
NUM_USERS = 1000
NUM_ITEMS = 2000
NUM_PTS = 50000
BATCH_SIZE = 2048
41
EVAL_BATCH_SIZE = 4000
42
43
NUM_NEG = 4

44
45
46
47
48
END_TO_END_TRAIN_MD5 = "b218738e915e825d03939c5e305a2698"
END_TO_END_EVAL_MD5 = "d753d0f3186831466d6e218163a9501e"
FRESH_RANDOMNESS_MD5 = "63d0dff73c0e5f1048fbdc8c65021e22"


49
50
51
def mock_download(*args, **kwargs):
  return

52

53
54
55
56
# The forkpool used by data producers interacts badly with the threading
# used by TestCase. Without this patch tests will hang, and no amount
# of diligent closing and joining within the producer will prevent it.
@mock.patch.object(popen_helper, "get_forkpool", popen_helper.get_fauxpool)
57
class BaseTest(tf.test.TestCase):
58

59
  def setUp(self):
60
    tf.compat.v1.disable_eager_execution()
61
62
    self.temp_data_dir = self.get_temp_dir()
    ratings_folder = os.path.join(self.temp_data_dir, DATASET)
63
    tf.io.gfile.makedirs(ratings_folder)
64
65
66
67
68
69
70
71
72
73
74
75
76
77
    np.random.seed(0)
    raw_user_ids = np.arange(NUM_USERS * 3)
    np.random.shuffle(raw_user_ids)
    raw_user_ids = raw_user_ids[:NUM_USERS]

    raw_item_ids = np.arange(NUM_ITEMS * 3)
    np.random.shuffle(raw_item_ids)
    raw_item_ids = raw_item_ids[:NUM_ITEMS]

    users = np.random.choice(raw_user_ids, NUM_PTS)
    items = np.random.choice(raw_item_ids, NUM_PTS)
    scores = np.random.randint(low=0, high=5, size=NUM_PTS)
    times = np.random.randint(low=1000000000, high=1200000000, size=NUM_PTS)

78
    self.rating_file = os.path.join(ratings_folder, movielens.RATINGS_FILE)
79
80
    self.seen_pairs = set()
    self.holdout = {}
81
    with tf.io.gfile.GFile(self.rating_file, "w") as f:
82
83
84
85
86
87
88
89
90
91
92
93
94
      f.write("user_id,item_id,rating,timestamp\n")
      for usr, itm, scr, ts in zip(users, items, scores, times):
        pair = (usr, itm)
        if pair in self.seen_pairs:
          continue
        self.seen_pairs.add(pair)
        if usr not in self.holdout or (ts, itm) > self.holdout[usr]:
          self.holdout[usr] = (ts, itm)

        f.write("{},{},{},{}\n".format(usr, itm, scr, ts))

    movielens.download = mock_download
    movielens.NUM_RATINGS[DATASET] = NUM_PTS
95
    movielens.DATASET_TO_NUM_USERS_AND_ITEMS[DATASET] = (NUM_USERS, NUM_ITEMS)
96

97
98
99
100
101
102
103
104
105
106
107
  def make_params(self, train_epochs=1):
    return {
        "train_epochs": train_epochs,
        "batches_per_step": 1,
        "use_seed": False,
        "batch_size": BATCH_SIZE,
        "eval_batch_size": EVAL_BATCH_SIZE,
        "num_neg": NUM_NEG,
        "match_mlperf": True,
        "use_tpu": False,
        "use_xla_for_gpu": False,
108
        "stream_files": False,
109
110
    }

111
112
  def test_preprocessing(self):
    # For the most part the necessary checks are performed within
113
114
    # _filter_index_sort()

115
116
117
118
119
120
    cache_path = os.path.join(self.temp_data_dir, "test_cache.pickle")
    data, valid_cache = data_preprocessing._filter_index_sort(
        self.rating_file, cache_path=cache_path)

    assert len(data[rconst.USER_MAP]) == NUM_USERS
    assert len(data[rconst.ITEM_MAP]) == NUM_ITEMS
121
122
123

  def drain_dataset(self, dataset, g):
    # type: (tf.data.Dataset, tf.Graph) -> list
124
    with self.session(graph=g) as sess:
125
      with g.as_default():
Jiri Simsa's avatar
Jiri Simsa committed
126
        batch = tf.compat.v1.data.make_one_shot_iterator(dataset).get_next()
127
128
129
130
131
132
133
134
      output = []
      while True:
        try:
          output.append(sess.run(batch))
        except tf.errors.OutOfRangeError:
          break
    return output

135
  def _test_end_to_end(self, constructor_type):
136
137
    params = self.make_params(train_epochs=1)
    _, _, producer = data_preprocessing.instantiate_pipeline(
Hongkun Yu's avatar
Hongkun Yu committed
138
139
140
141
142
        dataset=DATASET,
        data_dir=self.temp_data_dir,
        params=params,
        constructor_type=constructor_type,
        deterministic=True)
143
144
145
146

    producer.start()
    producer.join()
    assert producer._fatal_exception is None
147

148
149
150
151
152
153
    user_inv_map = {v: k for k, v in producer.user_map.items()}
    item_inv_map = {v: k for k, v in producer.item_map.items()}

    # ==========================================================================
    # == Training Data =========================================================
    # ==========================================================================
154
155
    g = tf.Graph()
    with g.as_default():
156
157
158
      input_fn = producer.make_input_fn(is_training=True)
      dataset = input_fn(params)

159
160
    first_epoch = self.drain_dataset(dataset=dataset, g=g)

161
    counts = defaultdict(int)
162
163
164
165
    train_examples = {
        True: set(),
        False: set(),
    }
166

167
    md5 = hashlib.md5()
168
    for features, labels in first_epoch:
169
      data_list = [
170
171
172
173
174
          features[movielens.USER_COLUMN].flatten(),
          features[movielens.ITEM_COLUMN].flatten(),
          features[rconst.VALID_POINT_MASK].flatten(),
          labels.flatten()
      ]
Taylor Robie's avatar
Taylor Robie committed
175
176
177
      for i in data_list:
        md5.update(i.tobytes())

178
      for u, i, v, l in zip(*data_list):
179
180
        if not v:
          continue  # ignore padding
181

182
183
184
185
186
187
188
        u_raw = user_inv_map[u]
        i_raw = item_inv_map[i]
        if ((u_raw, i_raw) in self.seen_pairs) != l:
          # The evaluation item is not considered during false negative
          # generation, so it will occasionally appear as a negative example
          # during training.
          assert not l
189
          self.assertEqual(i_raw, self.holdout[u_raw][1])
190
        train_examples[l].add((u_raw, i_raw))
191
192
        counts[(u_raw, i_raw)] += 1

193
    self.assertRegexpMatches(md5.hexdigest(), END_TO_END_TRAIN_MD5)
194

195
    num_positives_seen = len(train_examples[True])
196
    self.assertEqual(producer._train_pos_users.shape[0], num_positives_seen)
197
198
199

    # This check is more heuristic because negatives are sampled with
    # replacement. It only checks that negative generation is reasonably random.
200
201
202
203
204
205
206
207
208
209
210
211
212
213
    self.assertGreater(
        len(train_examples[False]) / NUM_NEG / num_positives_seen, 0.9)

    # This checks that the samples produced are independent by checking the
    # number of duplicate entries. If workers are not properly independent there
    # will be lots of repeated pairs.
    self.assertLess(np.mean(list(counts.values())), 1.1)

    # ==========================================================================
    # == Eval Data =============================================================
    # ==========================================================================
    with g.as_default():
      input_fn = producer.make_input_fn(is_training=False)
      dataset = input_fn(params)
214

215
    eval_data = self.drain_dataset(dataset=dataset, g=g)
216

217
    current_user = None
218
    md5 = hashlib.md5()
219
    for features in eval_data:
220
      data_list = [
221
222
223
224
          features[movielens.USER_COLUMN].flatten(),
          features[movielens.ITEM_COLUMN].flatten(),
          features[rconst.DUPLICATE_MASK].flatten()
      ]
Taylor Robie's avatar
Taylor Robie committed
225
226
227
      for i in data_list:
        md5.update(i.tobytes())

228
      for idx, (u, i, d) in enumerate(zip(*data_list)):
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
        u_raw = user_inv_map[u]
        i_raw = item_inv_map[i]
        if current_user is None:
          current_user = u

        # Ensure that users appear in blocks, as the evaluation logic expects
        # this structure.
        self.assertEqual(u, current_user)

        # The structure of evaluation data is 999 negative examples followed
        # by the holdout positive.
        if not (idx + 1) % (rconst.NUM_EVAL_NEGATIVES + 1):
          # Check that the last element in each chunk is the holdout item.
          self.assertEqual(i_raw, self.holdout[u_raw][1])
          current_user = None

        elif i_raw == self.holdout[u_raw][1]:
          # Because the holdout item is not given to the negative generation
          # process, it can appear as a negative. In that case, it should be
          # masked out as a duplicate. (Since the true positive is placed at
          # the end and would therefore lose the tie.)
          assert d

        else:
          # Otherwise check that the other 999 points for a user are selected
          # from the negatives.
          assert (u_raw, i_raw) not in self.seen_pairs

257
258
259
    self.assertRegexpMatches(md5.hexdigest(), END_TO_END_EVAL_MD5)

  def _test_fresh_randomness(self, constructor_type):
260
261
262
    train_epochs = 5
    params = self.make_params(train_epochs=train_epochs)
    _, _, producer = data_preprocessing.instantiate_pipeline(
Hongkun Yu's avatar
Hongkun Yu committed
263
264
265
266
267
        dataset=DATASET,
        data_dir=self.temp_data_dir,
        params=params,
        constructor_type=constructor_type,
        deterministic=True)
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282

    producer.start()

    results = []
    g = tf.Graph()
    with g.as_default():
      for _ in range(train_epochs):
        input_fn = producer.make_input_fn(is_training=True)
        dataset = input_fn(params)
        results.extend(self.drain_dataset(dataset=dataset, g=g))

    producer.join()
    assert producer._fatal_exception is None

    positive_counts, negative_counts = defaultdict(int), defaultdict(int)
283
    md5 = hashlib.md5()
284
    for features, labels in results:
285
      data_list = [
286
287
288
289
290
          features[movielens.USER_COLUMN].flatten(),
          features[movielens.ITEM_COLUMN].flatten(),
          features[rconst.VALID_POINT_MASK].flatten(),
          labels.flatten()
      ]
Taylor Robie's avatar
Taylor Robie committed
291
292
293
      for i in data_list:
        md5.update(i.tobytes())

294
      for u, i, v, l in zip(*data_list):
295
296
297
298
299
300
301
302
        if not v:
          continue  # ignore padding

        if l:
          positive_counts[(u, i)] += 1
        else:
          negative_counts[(u, i)] += 1

303
304
    self.assertRegexpMatches(md5.hexdigest(), FRESH_RANDOMNESS_MD5)

305
    # The positive examples should appear exactly once each epoch
Hongkun Yu's avatar
Hongkun Yu committed
306
307
    self.assertAllEqual(
        list(positive_counts.values()), [train_epochs for _ in positive_counts])
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324

    # The threshold for the negatives is heuristic, but in general repeats are
    # expected, but should not appear too frequently.

    pair_cardinality = NUM_USERS * NUM_ITEMS
    neg_pair_cardinality = pair_cardinality - len(self.seen_pairs)

    # Approximation for the expectation number of times that a particular
    # negative will appear in a given epoch. Implicit in this calculation is the
    # treatment of all negative pairs as equally likely. Normally is not
    # necessarily reasonable; however the generation in self.setUp() will
    # approximate this behavior sufficiently for heuristic testing.
    e_sample = len(self.seen_pairs) * NUM_NEG / neg_pair_cardinality

    # The frequency of occurance of a given negative pair should follow an
    # approximately binomial distribution in the limit that the cardinality of
    # the negative pair set >> number of samples per epoch.
Hongkun Yu's avatar
Hongkun Yu committed
325
326
    approx_pdf = scipy.stats.binom.pmf(
        k=np.arange(train_epochs + 1), n=train_epochs, p=e_sample)
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344

    # Tally the actual observed counts.
    count_distribution = [0 for _ in range(train_epochs + 1)]
    for i in negative_counts.values():
      i = min([i, train_epochs])  # round down tail for simplicity.
      count_distribution[i] += 1
    count_distribution[0] = neg_pair_cardinality - sum(count_distribution[1:])

    # Check that the frequency of negative pairs is approximately binomial.
    for i in range(train_epochs + 1):
      if approx_pdf[i] < 0.05:
        continue  # Variance will be high at the tails.

      observed_fraction = count_distribution[i] / neg_pair_cardinality
      deviation = (2 * abs(observed_fraction - approx_pdf[i]) /
                   (observed_fraction + approx_pdf[i]))

      self.assertLess(deviation, 0.2)
345

Taylor Robie's avatar
Taylor Robie committed
346
347
  def test_end_to_end_materialized(self):
    self._test_end_to_end("materialized")
348

Taylor Robie's avatar
Taylor Robie committed
349
350
351
352
353
354
355
356
  def test_end_to_end_bisection(self):
    self._test_end_to_end("bisection")

  def test_fresh_randomness_materialized(self):
    self._test_fresh_randomness("materialized")

  def test_fresh_randomness_bisection(self):
    self._test_fresh_randomness("bisection")
357

358
359

if __name__ == "__main__":
Taylor Robie's avatar
Taylor Robie committed
360
  tf.test.main()