Merge pull request #1791 from zhengya01/ce_ctr

Ce ctr

fluid/PaddleRec/ctr/.run_ce.sh

+#!/bin/bash
+
+export MKL_NUM_THREADS=1
+export OMP_NUM_THREADS=1
+
+
+#cudaid=${face_detection:=0} # use 0-th card as default
+#export CUDA_VISIBLE_DEVICES=$cudaid
+export CPU_NUM=1
+export NUM_THREADS=1
+
+FLAGS_benchmark=true  python train.py --is_local 1 --cloud_train 0 --train_data_path data/raw/train.txt --enable_ce | python _ce.py
+
+export CPU_NUM=1
+export NUM_THREADS=8
+
+FLAGS_benchmark=true  python train.py --is_local 1 --cloud_train 0 --train_data_path data/raw/train.txt --enable_ce | python _ce.py
+
+export CPU_NUM=8
+export NUM_THREADS=8
+
+FLAGS_benchmark=true  python train.py --is_local 1 --cloud_train 0 --train_data_path data/raw/train.txt --enable_ce | python _ce.py

fluid/PaddleRec/ctr/_ce.py

+# this file is only used for continuous evaluation test!
+
+import os
+import sys
+sys.path.append(os.environ['ceroot'])
+from kpi import CostKpi
+from kpi import DurationKpi
+from kpi import AccKpi
+
+
+each_pass_duration_cpu1_thread1_kpi = DurationKpi('each_pass_duration_cpu1_thread1', 0.08, 0, actived=True)
+train_loss_cpu1_thread1_kpi = CostKpi('train_loss_cpu1_thread1', 0.08, 0)
+train_auc_val_cpu1_thread1_kpi = AccKpi('train_auc_val_cpu1_thread1', 0.08, 0)
+train_batch_auc_val_cpu1_thread1_kpi = AccKpi('train_batch_auc_val_cpu1_thread1', 0.08, 0)
+each_pass_duration_cpu1_thread8_kpi = DurationKpi('each_pass_duration_cpu1_thread8', 0.08, 0, actived=True)
+train_loss_cpu1_thread8_kpi = CostKpi('train_loss_cpu1_thread8', 0.08, 0)
+train_auc_val_cpu1_thread8_kpi = AccKpi('train_auc_val_cpu1_thread8', 0.08, 0)
+train_batch_auc_val_cpu1_thread8_kpi = AccKpi('train_batch_auc_val_cpu1_thread8', 0.08, 0)
+each_pass_duration_cpu8_thread8_kpi = DurationKpi('each_pass_duration_cpu8_thread8', 0.08, 0, actived=True)
+train_loss_cpu8_thread8_kpi = CostKpi('train_loss_cpu8_thread8', 0.08, 0)
+train_auc_val_cpu8_thread8_kpi = AccKpi('train_auc_val_cpu8_thread8', 0.08, 0)
+train_batch_auc_val_cpu8_thread8_kpi = AccKpi('train_batch_auc_val_cpu8_thread8', 0.08, 0)
+
+tracking_kpis = [
+        each_pass_duration_cpu1_thread1_kpi,
+        train_loss_cpu1_thread1_kpi,
+        train_auc_val_cpu1_thread1_kpi,
+        train_batch_auc_val_cpu1_thread1_kpi,
+        each_pass_duration_cpu1_thread8_kpi,
+        train_loss_cpu1_thread8_kpi,
+        train_auc_val_cpu1_thread8_kpi,
+        train_batch_auc_val_cpu1_thread8_kpi,
+        each_pass_duration_cpu8_thread8_kpi,
+        train_loss_cpu8_thread8_kpi,
+        train_auc_val_cpu8_thread8_kpi,
+        train_batch_auc_val_cpu8_thread8_kpi,
+        ]
+
+
+def parse_log(log):
+    '''
+    This method should be implemented by model developers.
+
+    The suggestion:
+
+    each line in the log should be key, value, for example:
+
+    "
+    train_cost\t1.0
+    test_cost\t1.0
+    train_cost\t1.0
+    train_cost\t1.0
+    train_acc\t1.2
+    "
+    '''
+    for line in log.split('\n'):
+        fs = line.strip().split('\t')
+        print(fs)
+        if len(fs) == 3 and fs[0] == 'kpis':
+            kpi_name = fs[1]
+            kpi_value = float(fs[2])
+            yield kpi_name, kpi_value
+
+
+def log_to_ce(log):
+    kpi_tracker = {}
+    for kpi in tracking_kpis:
+        kpi_tracker[kpi.name] = kpi
+
+    for (kpi_name, kpi_value) in parse_log(log):
+        print(kpi_name, kpi_value)
+        kpi_tracker[kpi_name].add_record(kpi_value)
+        kpi_tracker[kpi_name].persist()
+
+
+if __name__ == '__main__':
+    log = sys.stdin.read()
+    log_to_ce(log)

fluid/PaddleRec/ctr/train.py

@@ -107,12 +107,22 @@ def parse_args():
         type=int,
         default=1,
         help='The num of trianers, (default: 1)')
+    parser.add_argument(
+        '--enable_ce',
+        action='store_true',
+        help='If set, run the task with continuous evaluation logs.')
 
     return parser.parse_args()
 
 
 def train_loop(args, train_program, py_reader, loss, auc_var, batch_auc_var,
                trainer_num, trainer_id):
+    
+    if args.enable_ce:
+        SEED = 102
+        train_program.random_seed = SEED
+        fluid.default_startup_program().random_seed = SEED
+
     dataset = reader.CriteoDataset(args.sparse_feature_dim)
     train_reader = paddle.batch(
         paddle.reader.shuffle(
@@ -146,6 +156,7 @@ def train_loop(args, train_program, py_reader, loss, auc_var, batch_auc_var,
 
     exe.run(fluid.default_startup_program())
 
+    total_time = 0
     for pass_id in range(args.num_passes):
         pass_start = time.time()
         batch_id = 0
@@ -169,10 +180,25 @@ def train_loop(args, train_program, py_reader, loss, auc_var, batch_auc_var,
             py_reader.reset()
         print("pass_id: %d, pass_time_cost: %f" % (pass_id, time.time() - pass_start))
 
+        total_time += time.time() - pass_start
+
         model_dir = args.model_output_dir + '/pass-' + str(pass_id)
         if args.trainer_id == 0:
             fluid.io.save_inference_model(model_dir, data_name_list, [loss, auc_var], exe)
 
+    # only for ce
+    if args.enable_ce:
+        threads_num, cpu_num = get_cards(args)
+        epoch_idx = args.num_passes 
+        print("kpis\teach_pass_duration_cpu%s_thread%s\t%s" %
+                (cpu_num, threads_num, total_time / epoch_idx))
+        print("kpis\ttrain_loss_cpu%s_thread%s\t%s" %
+                (cpu_num, threads_num, loss_val/args.batch_size))
+        print("kpis\ttrain_auc_val_cpu%s_thread%s\t%s" %
+                (cpu_num, threads_num, auc_val))
+        print("kpis\ttrain_batch_auc_val_cpu%s_thread%s\t%s" %
+                (cpu_num, threads_num, batch_auc_val))
+        
 
 def train():
     args = parse_args()
@@ -224,5 +250,11 @@ def train():
             )
 
 
+def get_cards(args):
+    threads_num = os.environ.get('NUM_THREADS', 1)
+    cpu_num = os.environ.get('CPU_NUM', 1)
+    return int(threads_num), int(cpu_num)
+
+
 if __name__ == '__main__':
     train()