Merge pull request #639 from esythan/new_trainer

New online trainer

Author: frankwhzhang

models/rank/slot_dnn/config_offline_infer.yaml

@@ -28,7 +28,8 @@ runner:
   reader_type: "InmemoryDataset"  # DataLoader / QueueDataset / RecDataset
   pipe_command: "python3 inmemorydataset_reader.py"
 
-  init_model_path: "output_model/20190720/inference_model_6"
+  init_model_path: "output_model/20190720/6"
+  model_mode: 0
 
   dataset_debug: False
   parse_ins_id: True
@@ -52,3 +53,44 @@ hyper_parameters:
   slot_num: 300
   layer_sizes: [512, 256, 128]
   distributed_embedding: 0
+
+table_parameters:
+  embedding:
+    table_class: "MemorySparseTable"
+    shard_num: 10
+    accessor:
+      accessor_class: "CtrCommonAccessor"
+      fea_dim: 11
+      embedx_dim: 8
+      embedx_threshold: 10
+      embed_sgd_param:
+        name: "SparseAdaGradSGDRule"
+        adagrad:
+          learning_rate: 0.05
+          initial_g2sum: 1.0
+          initial_range: 0.0001
+          weight_bounds: [-10.0, 10.0]
+      embedx_sgd_param:
+        name: "SparseAdaGradSGDRule"
+        adagrad:
+          learning_rate: 0.05
+          initial_g2sum: 3.0
+          initial_range: 0.0001
+          weight_bounds: [-10.0, 10.0]
+      ctr_accessor_param:
+        nonclk_coeff: 0.1
+        click_coeff: 1.0
+        # base_threshold: 1.5
+        # delta_threshold: 0.25
+        base_threshold: 0
+        delta_threshold: 0
+        delta_keep_days: 16.0
+        show_click_decay_rate: 0.98
+        delete_threshold: 0.8
+        delete_after_unseen_days: 30.0
+        ssd_unseenday_threshold: 1
+      # table_accessor_save_param:
+      #   num: 2
+      #   param: [1, 2]
+      #   converter: ""
+      #   deconverter: ""

models/rank/slot_dnn/config_online.yaml

@@ -2,47 +2,43 @@ runner:
   use_gloo: True
   # train_data_dir: "afs:/xxx"
   train_data_dir: "./data"
-  train_reader_path: "criteo_reader" # importlib format
   use_gpu: False
-  use_auc: True
   train_batch_size: 32
-  epochs: 3
   print_interval: 1
   model_save_path: "output_model"
-  checkpoint_per_pass: 6
-  save_delta_frequency: 6
+  # model_save_path: "afs:/user/paddle/wangguanqun/pscore_output_model"
+  checkpoint_per_pass: 1
+  save_delta_frequency: 1
 
   train_thread_num: 3
+  shuffle_thread_num: 12
   reader_type: "InMemoryDataset"  # DataLoader / QueueDataset / RecDataset
   pipe_command: "python3.7 queuedataset_reader.py"
   dataset_debug: False
-  split_file_list: False
   # data_donefile: "data.done"
   data_sleep_second: 1
   sync_mode: "async"
 
-  split_interval: 5
+  split_interval: 30
   split_per_pass: 2
   start_day: "20190720"
   end_day: "20190722"
   infer_batch_size: 32
   infer_thread_num: 1
-  infer_reader_path: "criteo_reader" # importlib format
   infer_data_dir: "data/"
-  infer_load_path: "output_model"
-  infer_start_epoch: 0
-  infer_end_epoch: 3
-  use_inference: True
 
   # need_train_dump: True
   # need_infer_dump: True
   train_dump_fields_dir: "./train_dump_data"
   infer_dump_fields_dir: "./infer_dump_data"
+
+  use_hadoop: False
   # fs_client:
   #   uri: "afs://xxx"
   #   user: "xxx"
   #   passwd: "xxx"
-  #   hadoop_bin: "hadoop"    
+  #   hadoop_bin: "$HADOOP_HOME/bin/hadoop"
+
 # hyper parameters of user-defined network
 hyper_parameters:
   # optimizer config
@@ -57,3 +53,43 @@ hyper_parameters:
   slot_num: 300
   layer_sizes: [512, 256, 128]
   distributed_embedding: 0
+  # adam_d2sum: False
+
+table_parameters:
+  embedding:
+    table_class: "MemorySparseTable"
+    shard_num: 10
+    accessor:
+      accessor_class: "CtrCommonAccessor"
+      fea_dim: 11
+      embedx_dim: 8
+      embedx_threshold: 10
+      embed_sgd_param:
+        name: "SparseAdaGradSGDRule"
+        adagrad:
+          learning_rate: 0.05
+          initial_g2sum: 3.0
+          initial_range: 0.0001
+          weight_bounds: [-10.0, 10.0]
+      embedx_sgd_param:
+        name: "SparseAdaGradSGDRule"
+        adagrad:
+          learning_rate: 0.05
+          initial_g2sum: 3.0
+          initial_range: 0.0001
+          weight_bounds: [-10.0, 10.0]
+      ctr_accessor_param:
+        nonclk_coeff: 0.1
+        click_coeff: 1.0
+        base_threshold: 1.5
+        delta_threshold: 0.25
+        delta_keep_days: 16.0
+        show_click_decay_rate: 0.98
+        delete_threshold: 0.8
+        delete_after_unseen_days: 30.0
+        ssd_unseenday_threshold: 1
+      # table_accessor_save_param:
+      #   num: 2
+      #   param: [1, 2]
+      #   converter: ""
+      #   deconverter: ""

models/rank/slot_dnn/inmemorydataset_reader.py

@@ -78,6 +78,9 @@ def line_process(self, line):
                 output[self.slot2index[i]][1].extend([self.padding])
             else:
                 self.visit[slot] = False
+
+        # add show
+        output = [("0", ["1"])] + output
         output = [("ins_id", [ins_id])] + output
         return output
         #return [label] + sparse_feature + [dense_feature]

models/rank/slot_dnn/net.py

@@ -16,6 +16,7 @@
 import paddle.nn as nn
 import paddle.nn.functional as F
 import math
+import paddle.fluid as fluid
 
 
 class BenchmarkDNNLayer(nn.Layer):
@@ -33,15 +34,7 @@ def __init__(self,
         self.layer_sizes = layer_sizes
         self._init_range = 0.2
 
-        # to do
-        #self.embedding = paddle.nn.Embedding(
-        #    self.dict_dim,
-        #    self.emb_dim,
-        #    sparse=True,
-        #    weight_attr=paddle.ParamAttr(
-        #        name="embedding",
-        #        initializer=paddle.nn.initializer.XavierNormal()))
-        #initializer=paddle.nn.initializer.Uniform()))
+        self.entry = paddle.distributed.ShowClickEntry("show", "click")
 
         sizes = [emb_dim * slot_num] + self.layer_sizes + [1]
         acts = ["relu" for _ in range(len(self.layer_sizes))] + [None]
@@ -70,7 +63,10 @@ def forward(self, slot_inputs):
             emb = paddle.static.nn.sparse_embedding(
                 input=s_input,
                 size=[self.dict_dim, self.emb_dim],
+                padding_idx=0,
+                entry=self.entry,
                 param_attr=paddle.ParamAttr(name="embedding"))
+
             self.inference_feed_vars.append(emb)
 
             bow = paddle.fluid.layers.sequence_pool(input=emb, pool_type='sum')

models/rank/slot_dnn/queuedataset_reader.py

@@ -78,8 +78,10 @@ def line_process(self, line):
             else:
                 self.visit[slot] = False
 
+        # add show
+        output = [("0", [1])] + output
         return output
-        #return [label] + sparse_feature + [dense_feature]
+
     def generate_sample(self, line):
         "Dataset Generator"
 

models/rank/slot_dnn/static_model.py

@@ -49,15 +49,17 @@ def create_feeds(self, is_infer=False):
             for i in range(2, self.slot_num + 2)
         ]
 
+        show = paddle.static.data(
+            name="show", shape=[None, 1], dtype="int64", lod_level=1)
         label = paddle.static.data(
             name="click", shape=[None, 1], dtype="int64", lod_level=1)
 
-        feeds_list = [label] + slot_ids
+        feeds_list = [show, label] + slot_ids
         return feeds_list
 
     def net(self, input, is_infer=False):
-        self.label_input = input[0]
-        self.slot_inputs = input[1:]
+        self.label_input = input[1]
+        self.slot_inputs = input[2:]
 
         dnn_model = BenchmarkDNNLayer(
             self.dict_dim,
@@ -74,12 +76,23 @@ def net(self, input, is_infer=False):
         predict_2d = paddle.concat(x=[1 - self.predict, self.predict], axis=1)
         #label_int = paddle.cast(self.label, 'int64')
 
-        auc, batch_auc_var, self.auc_stat_list = paddle.static.auc(
+        auc, batch_auc_var, auc_stat_list = paddle.static.auc(
             input=predict_2d, label=self.label_input, slide_steps=0)
-        self.metric_list = fluid.contrib.layers.ctr_metric_bundle(
+        metric_list = fluid.contrib.layers.ctr_metric_bundle(
             self.predict,
             fluid.layers.cast(
                 x=self.label_input, dtype='float32'))
+
+        self.thread_stat_var_names = [
+            auc_stat_list[2].name, auc_stat_list[3].name
+        ]
+        self.thread_stat_var_names += [i.name for i in metric_list]
+        self.thread_stat_var_names = list(set(self.thread_stat_var_names))
+
+        self.metric_list = list(auc_stat_list) + list(metric_list)
+        self.metric_types = ["int64"] * len(auc_stat_list) + ["float32"] * len(
+            metric_list)
+
         self.inference_feed_vars = dnn_model.inference_feed_vars
         self.inference_target_var = self.predict
 

tools/feature_importance.py

@@ -134,7 +134,9 @@ def run_offline_infer(self):
 
         self.exe.run(paddle.static.default_startup_program())
         fleet.init_worker()
-        fleet.load_model(init_model_path, mode=0)
+        if fleet.is_first_worker():
+            fleet.load_model(init_model_path, mode=0)
+        fleet.barrier_worker()
 
         logger.info("Prepare Dataset Begin.")
         prepare_data_start_time = time.time()

tools/static_ps_offline_infer.py

@@ -46,7 +46,7 @@ def parse_args():
     args = parser.parse_args()
     args.abs_dir = os.path.dirname(os.path.abspath(args.config_yaml))
     yaml_helper = YamlHelper()
-    config = yaml_helper.load_yaml(args.config_yaml)
+    config = yaml_helper.load_yaml(args.config_yaml, ["table_parameters"])
     config["yaml_path"] = args.config_yaml
     config["config_abs_dir"] = args.abs_dir
     yaml_helper.print_yaml(config)
@@ -121,7 +121,10 @@ def run_offline_infer(self):
         fleet.init_worker()
 
         init_model_path = config.get("runner.init_model_path")
-        fleet.load_model(init_model_path, mode=0)
+        model_mode = config.get("runner.model_mode", 0)
+        if fleet.is_first_worker():
+            fleet.load_model(init_model_path, mode=model_mode)
+        fleet.barrier_worker()
 
         logger.info("Prepare Dataset Begin.")
         prepare_data_start_time = time.time()
@@ -147,7 +150,6 @@ def dataset_offline_infer(self, cur_dataset):
             "dump_fields_path": dump_fields_path,
             "dump_fields": dump_fields
         })
-        print(paddle.static.default_main_program()._fleet_opt)
 
         self.exe.infer_from_dataset(
             program=paddle.static.default_main_program(),