esm pyspark 代码重构

tensnsorflow/feature_engineering.py → eda/esmm/Model_pipline/feature_engineering.py

@@ -206,7 +206,7 @@ def feature_engineer():
 
     # TODO 上线后把下面train fliter 删除，因为最近一天的数据也要作为训练集
 
-    train = rdd.filter(lambda x: x[0] != validate_date).map(
+    train = rdd.map(
         lambda x: (x[1], x[2], x[3], x[4], x[5], x[6], x[7], x[8], x[9],
                    x[10], x[11], x[12], x[13]))
     f = time.time()

eda/esmm/Model_pipline/submit.sh

 #! /bin/bash
 git checkout master
-PYTHON_PATH=/opt/anaconda3/envs/esmm/bin/python
-MODEL_PATH=/srv/apps/ffm-baseline/eda/esmm/Model_pipline
-DATA_PATH=/home/gmuser/esmm
-
-echo "rm leave tfrecord"
-rm ${DATA_PATH}/tr/*
-rm ${DATA_PATH}/va/*
-rm ${DATA_PATH}/native/*
-rm ${DATA_PATH}/nearby/*
-rm -r ${DATA_PATH}/model_ckpt/DeepCvrMTL/20*
-
-echo "data"
-${PYTHON_PATH} ${MODEL_PATH}/feature.py > ${DATA_PATH}/feature.log
-
-echo "csv to tfrecord"
-${PYTHON_PATH} ${MODEL_PATH}/to_tfrecord.py --input_dir=${DATA_PATH}/tr/ --output_dir=${DATA_PATH}/tr/
-${PYTHON_PATH} ${MODEL_PATH}/to_tfrecord.py --input_dir=${DATA_PATH}/va/ --output_dir=${DATA_PATH}/va/
-${PYTHON_PATH} ${MODEL_PATH}/to_tfrecord.py --input_dir=${DATA_PATH}/native/ --output_dir=${DATA_PATH}/native/
-${PYTHON_PATH} ${MODEL_PATH}/to_tfrecord.py --input_dir=${DATA_PATH}/nearby/ --output_dir=${DATA_PATH}/nearby/
-
-cat ${DATA_PATH}/tr/*.tfrecord > ${DATA_PATH}/tr/tr.tfrecord
-cat ${DATA_PATH}/va/*.tfrecord > ${DATA_PATH}/va/va.tfrecord
-cat ${DATA_PATH}/native/*.tfrecord > ${DATA_PATH}/native/native.tfrecord
-cat ${DATA_PATH}/nearby/*.tfrecord > ${DATA_PATH}/nearby/nearby.tfrecord
-
-rm ${DATA_PATH}/tr/tr_*
-rm ${DATA_PATH}/va/va_*
-rm ${DATA_PATH}/native/native_*
-rm ${DATA_PATH}/nearby/nearby_*
+PYTHON_PATH=/srv/envs/esmm/bin/python
+MODEL_PATH=/srv/apps/ffm-baseline_git/eda/esmm/Model_pipline
+LOCAL_PATH=/home/gmuser/esmm
+HDFS_PATH=hdfs://172.16.32.4:8020/strategy/esmm
 
+echo "rm model file"
+rm -r ${LOCAL_PATH}/model_ckpt/DeepCvrMTL/20*
 
 echo "train..."
-${PYTHON_PATH} ${MODEL_PATH}/train.py --ctr_task_wgt=0.5 --learning_rate=0.0001 --deep_layers=512,256,128,64,32 --dropout=0.3,0.3,0.3,0.3,0.3 --optimizer=Adam --num_epochs=1 --embedding_size=16 --batch_size=2000 --field_size=15 --feature_size=300000 --l2_reg=0.005 --log_steps=100 --num_threads=36 --model_dir=${DATA_PATH}/model_ckpt/DeepCvrMTL/ --data_dir=${DATA_PATH} --task_type=train
-
+CLASSPATH="$(hadoop classpath --glob)" ${PYTHON_PATH} ${MODEL_PATH}/train.py --ctr_task_wgt=0.5 --learning_rate=0.0001 --deep_layers=512,256,128,64,32 --dropout=0.3,0.3,0.3,0.3,0.3 --optimizer=Adam --num_epochs=1 --embedding_size=16 --batch_size=10000 --field_size=15 --feature_size=600000 --l2_reg=0.005 --log_steps=100 --num_threads=36 --model_dir=${LOCAL_PATH}/model_ckpt/DeepCvrMTL/ --local_dir=${LOCAL_PATH} --task_type=train
 
 echo "infer native..."
-${PYTHON_PATH} ${MODEL_PATH}/train.py --ctr_task_wgt=0.5 --learning_rate=0.0001 --deep_layers=512,256,128,64,32 --dropout=0.3,0.3,0.3,0.3,0.3 --optimizer=Adam --num_epochs=1 --embedding_size=16 --batch_size=2000 --field_size=15 --feature_size=300000 --l2_reg=0.005 --log_steps=100 --num_threads=36 --model_dir=${DATA_PATH}/model_ckpt/DeepCvrMTL/ --data_dir=${DATA_PATH}/native --task_type=infer
+CLASSPATH="$(hadoop classpath --glob)" ${PYTHON_PATH} ${MODEL_PATH}/train.py --ctr_task_wgt=0.5 --learning_rate=0.0001 --deep_layers=512,256,128,64,32 --dropout=0.3,0.3,0.3,0.3,0.3 --optimizer=Adam --num_epochs=1 --embedding_size=16 --batch_size=10000 --field_size=15 --feature_size=600000 --l2_reg=0.005 --log_steps=100 --num_threads=36 --model_dir=${LOCAL_PATH}/model_ckpt/DeepCvrMTL/ --local_dir=${LOCAL_PATH}/native --hdfs_dir=${HDFS_PATH}/native --task_type=infer
 
 
 echo "infer nearby..."
-${PYTHON_PATH} ${MODEL_PATH}/train.py --ctr_task_wgt=0.5 --learning_rate=0.0001 --deep_layers=512,256,128,64,32 --dropout=0.3,0.3,0.3,0.3,0.3 --optimizer=Adam --num_epochs=1 --embedding_size=16 --batch_size=2000 --field_size=15 --feature_size=300000 --l2_reg=0.005 --log_steps=100 --num_threads=36 --model_dir=${DATA_PATH}/model_ckpt/DeepCvrMTL/ --data_dir=${DATA_PATH}/nearby --task_type=infer
+CLASSPATH="$(hadoop classpath --glob)" ${PYTHON_PATH} ${MODEL_PATH}/train.py --ctr_task_wgt=0.5 --learning_rate=0.0001 --deep_layers=512,256,128,64,32 --dropout=0.3,0.3,0.3,0.3,0.3 --optimizer=Adam --num_epochs=1 --embedding_size=16 --batch_size=10000 --field_size=15 --feature_size=600000 --l2_reg=0.005 --log_steps=100 --num_threads=36 --model_dir=${LOCAL_PATH}/model_ckpt/DeepCvrMTL/ --local_dir=${LOCAL_PATH}/nearby --hdfs_dir=${HDFS_PATH}/nearby --task_type=infer
 
+echo "sort and 2sql"
+${PYTHON_PATH} ${MODEL_PATH}/to_database.py
\ No newline at end of file

eda/esmm/Model_pipline/to_database.py

@@ -81,7 +81,6 @@ def main():
         tmp = str(to_delete[start:end]).strip('[]')
         df_merge_str.append(tmp)
 
-    try:
     for i in df_merge_str:
         delete_str = 'delete from esmm_device_diary_queue where concat(device_id,city_id) in ({0})'.format(i)
         con = pymysql.connect(host='172.16.40.158', port=4000, user='root', passwd='3SYz54LS9#^9sBvC', db='jerry_test')
@@ -91,11 +90,24 @@ def main():
         print("delete done")
         con.close()
     engine = create_engine(str(r"mysql+pymysql://%s:" + '%s' + "@%s:%s/%s") % (user, password, host, port, db))
-        df_all.to_sql('esmm_device_diary_queue',con=engine,if_exists='append',index=False,chunksize=8000)
+    df_all.to_sql('esmm_device_diary_queue', con=engine, if_exists='append', index=False, chunksize=8000)
     print("insert done")
 
-    except Exception as e:
-        print(e)
+    # try:
+    #     for i in df_merge_str:
+    #         delete_str = 'delete from esmm_device_diary_queue where concat(device_id,city_id) in ({0})'.format(i)
+    #         con = pymysql.connect(host='172.16.40.158', port=4000, user='root', passwd='3SYz54LS9#^9sBvC', db='jerry_test')
+    #         cur = con.cursor()
+    #         cur.execute(delete_str)
+    #         con.commit()
+    #         print("delete done")
+    #         con.close()
+    #     engine = create_engine(str(r"mysql+pymysql://%s:" + '%s' + "@%s:%s/%s") % (user, password, host, port, db))
+    #     df_all.to_sql('esmm_device_diary_queue',con=engine,if_exists='append',index=False,chunksize=8000)
+    #     print("insert done")
+    #
+    # except Exception as e:
+    #     print(e)
 
 
 if __name__ == '__main__':

tensnsorflow/es/to_database.py

-#coding=utf-8
-
-from sqlalchemy import create_engine
-import pandas as pd
-import pymysql
-import time
-
-def con_sql(sql):
-    """
-    :type sql : str
-    :rtype : tuple
-    """
-    db = pymysql.connect(host='10.66.157.22', port=4000, user='root', passwd='3SYz54LS9#^9sBvC', db='jerry_test')
-    cursor = db.cursor()
-    cursor.execute(sql)
-    result = cursor.fetchall()
-    db.close()
-    return result
-
-
-def nearby_set_join(lst):
-    # return ','.join([str(i) for i in list(lst)])
-    return ','.join([str(i) for i in lst.unique().tolist()])
-
-
-def native_set_join(lst):
-    l = lst.unique().tolist()
-    d = int(len(l)/2)
-    if d == 0:
-        d = 1
-    r = [str(i) for i in l]
-    r =r[:d]
-    return ','.join(r)
-
-
-def main():
-
-    # native queue
-    df2 = pd.read_csv('/data/esmm/native.csv')
-    df2['cid_id'] = df2['cid_id'].astype(str)
-
-    df1 = pd.read_csv("/data/esmm/native/pred.txt",sep='\t',header=None,names=["ctr","cvr","ctcvr"])
-    df2["ctr"],df2["cvr"],df2["ctcvr"] = df1["ctr"],df1["cvr"],df1["ctcvr"]
-    df3 = df2.groupby(by=["uid","city"]).apply(lambda x: x.sort_values(by="ctcvr",ascending=False)).reset_index(drop=True).groupby(by=["uid","city"]).agg({'cid_id':native_set_join}).reset_index(drop=False)
-    df3.columns = ["device_id","city_id","native_queue"]
-    print("native_device_count",df3.shape)
-
-
-    # nearby queue
-    df2 = pd.read_csv('/data/esmm/nearby.csv')
-    df2['cid_id'] = df2['cid_id'].astype(str)
-
-    df1 = pd.read_csv("/data/esmm/nearby/pred.txt",sep='\t',header=None,names=["ctr","cvr","ctcvr"])
-    df2["ctr"], df2["cvr"], df2["ctcvr"] = df1["ctr"], df1["cvr"], df1["ctcvr"]
-    df4 = df2.groupby(by=["uid","city"]).apply(lambda x: x.sort_values(by="ctcvr",ascending=False)).reset_index(drop=True).groupby(by=["uid","city"]).agg({'cid_id':nearby_set_join}).reset_index(drop=False)
-    df4.columns = ["device_id","city_id","nearby_queue"]
-    print("nearby_device_count",df4.shape)
-
-    #union
-    df_all = pd.merge(df3,df4,on=['device_id','city_id'],how='outer').fillna("")
-    df_all['device_id'] = df_all['device_id'].astype(str)
-    df_all['city_id'] = df_all['city_id'].astype(str)
-    ctime = int(time.time())
-    df_all["time"] = ctime
-    print("union_device_count",df_all.shape)
-
-    host='10.66.157.22'
-    port=4000
-    user='root'
-    password='3SYz54LS9#^9sBvC'
-    db='jerry_test'
-    charset='utf8'
-
-    engine = create_engine(str(r"mysql+mysqldb://%s:" + '%s' + "@%s:%s/%s") % (user, password, host, port, db))
-    df_merge = df_all['device_id'] + df_all['city_id']
-    df_merge_str = (str(list(df_merge.values))).strip('[]')
-    try:
-        # df_merge = df_all[['device_id','city_id']].apply(lambda x: ''.join(x),axis=1)
-
-        delete_str = 'delete from esmm_device_diary_queue where concat(device_id,city_id) in ({0})'.format(df_merge_str)
-        con = pymysql.connect(host='10.66.157.22', port=4000, user='root', passwd='3SYz54LS9#^9sBvC', db='jerry_test')
-        cur = con.cursor()
-        cur.execute(delete_str)
-        con.commit()
-        df_all.to_sql('esmm_device_diary_queue',con=engine,if_exists='append',index=False,chunksize=8000)
-    except Exception as e:
-        print(e)
-
-    print("done")
-
-if __name__ == '__main__':
-    main()
\ No newline at end of file

tensnsorflow/es/to_tfrecord.py

-#coding=utf-8
-
-from __future__ import absolute_import
-from __future__ import division
-from __future__ import print_function
-import pandas as pd
-import os
-import glob
-
-import tensorflow as tf
-import numpy as np
-from multiprocessing import Pool as ThreadPool
-
-flags = tf.app.flags
-FLAGS = flags.FLAGS
-LOG = tf.logging
-
-tf.app.flags.DEFINE_string("input_dir", "./", "input dir")
-tf.app.flags.DEFINE_string("output_dir", "./", "output dir")
-tf.app.flags.DEFINE_integer("threads", 16, "threads num")
-
-def gen_tfrecords(in_file):
-    basename = os.path.basename(in_file) + ".tfrecord"
-    out_file = os.path.join(FLAGS.output_dir, basename)
-    tfrecord_out = tf.python_io.TFRecordWriter(out_file)
-    df = pd.read_csv(in_file)
-
-    for i in range(df.shape[0]):
-        feats = ["ucity_id", "ccity_name", "device_type", "manufacturer",
-                "channel", "top", "time", "stat_date","hospital_id",
-                 "method", "min", "max", "treatment_time", "maintain_time", "recover_time"]
-        id = np.array([])
-        for j in feats:
-            id = np.append(id,df[j][i])
-        app_list = np.array(str(df["app_list"][i]).split(","))
-        level2_list = np.array(str(df["clevel2_id"][i]).split(","))
-        level3_list = np.array(str(df["level3_ids"][i]).split(","))
-        features = tf.train.Features(feature={
-            "y": tf.train.Feature(float_list=tf.train.FloatList(value=[df["y"][i]])),
-            "z": tf.train.Feature(float_list=tf.train.FloatList(value=[df["z"][i]])),
-            "ids": tf.train.Feature(int64_list=tf.train.Int64List(value=id.astype(np.int))),
-            "app_list":tf.train.Feature(int64_list=tf.train.Int64List(value=app_list.astype(np.int))),
-            "level2_list": tf.train.Feature(int64_list=tf.train.Int64List(value=level2_list.astype(np.int))),
-            "level3_list": tf.train.Feature(int64_list=tf.train.Int64List(value=level3_list.astype(np.int)))
-            })
-
-        example = tf.train.Example(features = features)
-        serialized = example.SerializeToString()
-        tfrecord_out.write(serialized)
-    tfrecord_out.close()
-
-def main(_):
-    if not os.path.exists(FLAGS.output_dir):
-        os.mkdir(FLAGS.output_dir)
-    file_list = glob.glob(os.path.join(FLAGS.input_dir, "*.csv"))
-    print("total files: %d" % len(file_list))
-
-    pool = ThreadPool(FLAGS.threads) # Sets the pool size
-    pool.map(gen_tfrecords, file_list)
-    pool.close()
-    pool.join()
-
-
-if __name__ == "__main__":
-    tf.logging.set_verbosity(tf.logging.INFO)
-    tf.app.run()
\ No newline at end of file

tensnsorflow/es/submit.sh → tensnsorflow/old_version/submit.sh

 #! /bin/bash
 git checkout master
-PYTHON_PATH=/home/gaoyazhe/miniconda3/bin/python
-MODEL_PATH=/srv/apps/ffm-baseline/tensnsorflow/es
-DATA_PATH=/data/esmm
+PYTHON_PATH=/opt/anaconda3/envs/esmm/bin/python
+MODEL_PATH=/srv/apps/ffm-baseline/eda/esmm/Model_pipline
+DATA_PATH=/home/gmuser/esmm
 
 echo "rm leave tfrecord"
 rm ${DATA_PATH}/tr/*
@@ -32,15 +32,13 @@ rm ${DATA_PATH}/nearby/nearby_*
 
 
 echo "train..."
-${PYTHON_PATH} ${MODEL_PATH}/train.py --ctr_task_wgt=0.5 --learning_rate=0.0001 --deep_layers=512,256,128,64,32 --dropout=0.3,0.3,0.3,0.3,0.3 --optimizer=Adam --num_epochs=1 --embedding_size=16 --batch_size=1024 --field_size=15 --feature_size=300000 --l2_reg=0.005 --log_steps=100 --num_threads=36 --model_dir=${DATA_PATH}/model_ckpt/DeepCvrMTL/ --data_dir=${DATA_PATH} --task_type=train
+${PYTHON_PATH} ${MODEL_PATH}/train.py --ctr_task_wgt=0.5 --learning_rate=0.0001 --deep_layers=512,256,128,64,32 --dropout=0.3,0.3,0.3,0.3,0.3 --optimizer=Adam --num_epochs=1 --embedding_size=16 --batch_size=2000 --field_size=15 --feature_size=300000 --l2_reg=0.005 --log_steps=100 --num_threads=36 --model_dir=${DATA_PATH}/model_ckpt/DeepCvrMTL/ --data_dir=${DATA_PATH} --task_type=train
 
 
 echo "infer native..."
-${PYTHON_PATH} ${MODEL_PATH}/train.py --ctr_task_wgt=0.5 --learning_rate=0.0001 --deep_layers=512,256,128,64,32 --dropout=0.3,0.3,0.3,0.3,0.3 --optimizer=Adam --num_epochs=1 --embedding_size=16 --batch_size=1024 --field_size=15 --feature_size=300000 --l2_reg=0.005 --log_steps=100 --num_threads=36 --model_dir=${DATA_PATH}/model_ckpt/DeepCvrMTL/ --data_dir=${DATA_PATH}/native --task_type=infer > ${DATA_PATH}/native_infer.log
+${PYTHON_PATH} ${MODEL_PATH}/train.py --ctr_task_wgt=0.5 --learning_rate=0.0001 --deep_layers=512,256,128,64,32 --dropout=0.3,0.3,0.3,0.3,0.3 --optimizer=Adam --num_epochs=1 --embedding_size=16 --batch_size=2000 --field_size=15 --feature_size=300000 --l2_reg=0.005 --log_steps=100 --num_threads=36 --model_dir=${DATA_PATH}/model_ckpt/DeepCvrMTL/ --data_dir=${DATA_PATH}/native --task_type=infer
 
 
 echo "infer nearby..."
-${PYTHON_PATH} ${MODEL_PATH}/train.py --ctr_task_wgt=0.5 --learning_rate=0.0001 --deep_layers=512,256,128,64,32 --dropout=0.3,0.3,0.3,0.3,0.3 --optimizer=Adam --num_epochs=1 --embedding_size=16 --batch_size=1024 --field_size=15 --feature_size=300000 --l2_reg=0.005 --log_steps=100 --num_threads=36 --model_dir=${DATA_PATH}/model_ckpt/DeepCvrMTL/ --data_dir=${DATA_PATH}/nearby --task_type=infer > ${DATA_PATH}/nearby_infer.log
+${PYTHON_PATH} ${MODEL_PATH}/train.py --ctr_task_wgt=0.5 --learning_rate=0.0001 --deep_layers=512,256,128,64,32 --dropout=0.3,0.3,0.3,0.3,0.3 --optimizer=Adam --num_epochs=1 --embedding_size=16 --batch_size=2000 --field_size=15 --feature_size=300000 --l2_reg=0.005 --log_steps=100 --num_threads=36 --model_dir=${DATA_PATH}/model_ckpt/DeepCvrMTL/ --data_dir=${DATA_PATH}/nearby --task_type=infer
 
-echo "sort and 2sql"
-${PYTHON_PATH} ${MODEL_PATH}/to_database.py > ${DATA_PATH}/insert_database.log

tensnsorflow/to_database.py → tensnsorflow/old_version/to_database.py

@@ -81,6 +81,7 @@ def main():
         tmp = str(to_delete[start:end]).strip('[]')
         df_merge_str.append(tmp)
 
+    try:
         for i in df_merge_str:
             delete_str = 'delete from esmm_device_diary_queue where concat(device_id,city_id) in ({0})'.format(i)
             con = pymysql.connect(host='172.16.40.158', port=4000, user='root', passwd='3SYz54LS9#^9sBvC', db='jerry_test')
@@ -90,24 +91,11 @@ def main():
             print("delete done")
             con.close()
         engine = create_engine(str(r"mysql+pymysql://%s:" + '%s' + "@%s:%s/%s") % (user, password, host, port, db))
-    df_all.to_sql('esmm_device_diary_queue', con=engine, if_exists='append', index=False, chunksize=8000)
+        df_all.to_sql('esmm_device_diary_queue',con=engine,if_exists='append',index=False,chunksize=8000)
         print("insert done")
 
-    # try:
-    #     for i in df_merge_str:
-    #         delete_str = 'delete from esmm_device_diary_queue where concat(device_id,city_id) in ({0})'.format(i)
-    #         con = pymysql.connect(host='172.16.40.158', port=4000, user='root', passwd='3SYz54LS9#^9sBvC', db='jerry_test')
-    #         cur = con.cursor()
-    #         cur.execute(delete_str)
-    #         con.commit()
-    #         print("delete done")
-    #         con.close()
-    #     engine = create_engine(str(r"mysql+pymysql://%s:" + '%s' + "@%s:%s/%s") % (user, password, host, port, db))
-    #     df_all.to_sql('esmm_device_diary_queue',con=engine,if_exists='append',index=False,chunksize=8000)
-    #     print("insert done")
-    #
-    # except Exception as e:
-    #     print(e)
+    except Exception as e:
+        print(e)
 
 
 if __name__ == '__main__':

tensnsorflow/es/train.py → tensnsorflow/old_version/train.py

@@ -6,7 +6,6 @@
 
 #import argparse
 import shutil
-#import sys
 import os
 import json
 import glob
@@ -55,7 +54,14 @@ def input_fn(filenames, batch_size=32, num_epochs=1, perform_shuffle=False):
             "ids": tf.FixedLenFeature([FLAGS.field_size], tf.int64),
             "app_list": tf.VarLenFeature(tf.int64),
             "level2_list": tf.VarLenFeature(tf.int64),
-            "level3_list": tf.VarLenFeature(tf.int64)
+            "level3_list": tf.VarLenFeature(tf.int64),
+            "tag1_list": tf.VarLenFeature(tf.int64),
+            "tag2_list": tf.VarLenFeature(tf.int64),
+            "tag3_list": tf.VarLenFeature(tf.int64),
+            "tag4_list": tf.VarLenFeature(tf.int64),
+            "tag5_list": tf.VarLenFeature(tf.int64),
+            "tag6_list": tf.VarLenFeature(tf.int64),
+            "tag7_list": tf.VarLenFeature(tf.int64)
         }
 
         parsed = tf.parse_single_example(record, features)
@@ -103,6 +109,14 @@ def model_fn(features, labels, mode, params):
     feat_ids = features['ids']
     app_list = features['app_list']
     level2_list = features['level2_list']
+    level3_list = features['level3_list']
+    tag1_list = features['tag1_list']
+    tag2_list = features['tag2_list']
+    tag3_list = features['tag3_list']
+    tag4_list = features['tag4_list']
+    tag5_list = features['tag5_list']
+    tag6_list = features['tag6_list']
+    tag7_list = features['tag7_list']
 
     if FLAGS.task_type != "infer":
         y = labels['y']
@@ -113,10 +127,18 @@ def model_fn(features, labels, mode, params):
         embedding_id = tf.nn.embedding_lookup(Feat_Emb,feat_ids)
         app_id = tf.nn.embedding_lookup_sparse(Feat_Emb, sp_ids=app_list, sp_weights=None, combiner="sum")
         level2 = tf.nn.embedding_lookup_sparse(Feat_Emb, sp_ids=level2_list, sp_weights=None, combiner="sum")
-
+        level3 = tf.nn.embedding_lookup_sparse(Feat_Emb, sp_ids=level3_list, sp_weights=None, combiner="sum")
+        tag1 = tf.nn.embedding_lookup_sparse(Feat_Emb, sp_ids=tag1_list, sp_weights=None, combiner="sum")
+        tag2 = tf.nn.embedding_lookup_sparse(Feat_Emb, sp_ids=tag2_list, sp_weights=None, combiner="sum")
+        tag3 = tf.nn.embedding_lookup_sparse(Feat_Emb, sp_ids=tag3_list, sp_weights=None, combiner="sum")
+        tag4 = tf.nn.embedding_lookup_sparse(Feat_Emb, sp_ids=tag4_list, sp_weights=None, combiner="sum")
+        tag5 = tf.nn.embedding_lookup_sparse(Feat_Emb, sp_ids=tag5_list, sp_weights=None, combiner="sum")
+        tag6 = tf.nn.embedding_lookup_sparse(Feat_Emb, sp_ids=tag6_list, sp_weights=None, combiner="sum")
+        tag7 = tf.nn.embedding_lookup_sparse(Feat_Emb, sp_ids=tag7_list, sp_weights=None, combiner="sum")
 
         # x_concat = tf.reshape(embedding_id,shape=[-1, common_dims])  # None * (F * K)
-        x_concat = tf.concat([tf.reshape(embedding_id,shape=[-1,common_dims]),app_id,level2], axis=1)
+        x_concat = tf.concat([tf.reshape(embedding_id,shape=[-1,common_dims]),app_id,level2,level3,tag1,
+                              tag2,tag3,tag4,tag5,tag6,tag7], axis=1)
 
     with tf.name_scope("CVR_Task"):
         if mode == tf.estimator.ModeKeys.TRAIN:

tensnsorflow/pipe.sh

-#! /bin/bash
-git checkout master
-PYTHON_PATH=/srv/envs/esmm/bin/python
-MODEL_PATH=/srv/apps/ffm-baseline_git/tensnsorflow
-LOCAL_PATH=/home/gmuser/esmm
-HDFS_PATH=hdfs://172.16.32.4:8020/strategy/esmm
-
-echo "train..."
-CLASSPATH="$(hadoop classpath --glob)" ${PYTHON_PATH} ${MODEL_PATH}/train_multi.py --ctr_task_wgt=0.5 --learning_rate=0.0001 --deep_layers=512,256,128,64,32 --dropout=0.3,0.3,0.3,0.3,0.3 --optimizer=Adam --num_epochs=1 --embedding_size=16 --batch_size=10000 --field_size=15 --feature_size=600000 --l2_reg=0.005 --log_steps=100 --num_threads=36 --model_dir=${LOCAL_PATH}/model_ckpt/DeepCvrMTL/ --local_dir=${LOCAL_PATH} --task_type=train
-
-echo "infer native..."
-CLASSPATH="$(hadoop classpath --glob)" ${PYTHON_PATH} ${MODEL_PATH}/train_multi.py --ctr_task_wgt=0.5 --learning_rate=0.0001 --deep_layers=512,256,128,64,32 --dropout=0.3,0.3,0.3,0.3,0.3 --optimizer=Adam --num_epochs=1 --embedding_size=16 --batch_size=10000 --field_size=15 --feature_size=600000 --l2_reg=0.005 --log_steps=100 --num_threads=36 --model_dir=${LOCAL_PATH}/model_ckpt/DeepCvrMTL/ --local_dir=${LOCAL_PATH}/native --hdfs_dir=${HDFS_PATH}/native --task_type=infer
-
-
-echo "infer nearby..."
-CLASSPATH="$(hadoop classpath --glob)" ${PYTHON_PATH} ${MODEL_PATH}/train_multi.py --ctr_task_wgt=0.5 --learning_rate=0.0001 --deep_layers=512,256,128,64,32 --dropout=0.3,0.3,0.3,0.3,0.3 --optimizer=Adam --num_epochs=1 --embedding_size=16 --batch_size=10000 --field_size=15 --feature_size=600000 --l2_reg=0.005 --log_steps=100 --num_threads=36 --model_dir=${LOCAL_PATH}/model_ckpt/DeepCvrMTL/ --local_dir=${LOCAL_PATH}/nearby --hdfs_dir=${HDFS_PATH}/nearby --task_type=infer
-
-echo "sort and 2sql"
-${PYTHON_PATH} ${MODEL_PATH}/to_database.py
\ No newline at end of file