Deep_match_rank_emb/match_recall/Dssm_tzld_match.py

# encoding: utf-8
import pandas as pd
import os
import gc
import time
from deepctr.feature_column import SparseFeat, VarLenSparseFeat, get_feature_names
# from preprocess_tzld210322_gen import gen_data_set, gen_model_input, gen_model_input_user_emb
# from preprocess_tzld210327_gen import gen_data_set, gen_model_input, gen_model_input_user_emb
from preprocess_tzld210423_gen import gen_data_set, gen_model_input, gen_model_input_user_emb

from sklearn.preprocessing import LabelEncoder
from tensorflow.python.keras import backend as K
from tensorflow.python.keras.models import Model
import tensorflow as tf
import numpy as np

from deepmatch.models import *

count_train = 1
count_test = 1

batch_size = 4096

def generate_arrays_from_train(train_set, user_profile, SEQ_LEN):
    # x_y 是我们的训练集包括标签，每一行的第一个是我们的图片路径，后面的是我们的独热化后的标签
    while 1:
            for i in range(0, len(train_set), batch_size):
                try:
                    train_batch = train_set[i: i + batch_size]
                    train_model_input_batch, train_label_batch = gen_model_input(train_batch, user_profile, SEQ_LEN)
                    if train_model_input_batch is None or train_label_batch is None:
                        continue

                    print("train i:  " + str(i) + "     len train set  " + str(len(train_set)))
                    print(train_model_input_batch)
                    print(train_label_batch)
                    yield (train_model_input_batch, train_label_batch)
                except Exception as ex:
                    print("\n\n generate_arrays_from_train exception ", ex)
                    continue

def generate_arrays_from_train_bak(train_set, user_profile, SEQ_LEN):
    global count_train
    while True:
        try:
            train_batch = train_set[(count_train - 1) * batch_size: count_train * batch_size]
            train_model_input_batch, train_label_batch = gen_model_input(train_batch, user_profile, SEQ_LEN)
            if count_train % 1000 == 0:
                print("count:" + str(count_train) + "     len train set  " + str(len(train_set)))
            count_train = count_train + 1
            if count_train * batch_size > len(train_set):
                count_train = 1
            yield (train_model_input_batch, train_label_batch)
        except Exception as ex:
            print("\n\n generate_arrays_from_file exception ", ex)
            count_train = count_train + 1
            continue


def generate_arrays_from_test(train_set, user_profile, SEQ_LEN):
    # x_y 是我们的训练集包括标签，每一行的第一个是我们的图片路径，后面的是我们的独热化后的标签

    while 1:
        for i in range(0, len(train_set), batch_size):
            try:
                train_batch = train_set[i: i + batch_size]
                # train_model_input_batch, train_label_batch = gen_model_input(train_batch, user_profile, SEQ_LEN)
                train_model_input_batch, train_label_batch = gen_model_input_user_emb(train_batch, user_profile, SEQ_LEN)
                if train_model_input_batch is None or train_label_batch is None:
                    continue
                print("test i:  " + str(i) + "     len train set  " + str(len(train_set)))
                yield (train_model_input_batch, train_label_batch)
            except Exception as ex:
                print("\n\n generate_arrays_from_test exception ", ex)
                continue


def generate_arrays_from_test_bak(train_set, user_profile, SEQ_LEN):
    global count_test
    while True:
        try:
            train_batch = train_set[(count_test - 1) * batch_size: count_test * batch_size]
            train_model_input_batch, train_label_batch = gen_model_input_user_emb(train_batch, user_profile, SEQ_LEN)
            if count_test % 1000 == 0:
                print("count:" + str(count_test) + "     len train set  " + str(len(train_set)))
            count_test = count_test + 1
            if count_test * batch_size > len(train_set):
                count_test = 1
            yield (train_model_input_batch, train_label_batch)
        except Exception as ex:
            print("\n\n generate_arrays_from_file exception ", ex)
            count_test = count_test + 1
            continue


if __name__ == "__main__":

    begin_time = time.time()

    data = pd.read_csvdata = pd.read_csv("/work/xielixun/user_action_feature_wdl_recall_app20210422.csv")
    print(data[0:5])
    sparse_features = ["videoid", "mid", "mobileos",
                       "videoGenre1", "videoGenre2", "userRatedVideo1", "userRatedVideo2", "userRatedVideo3", "userGenre1", "userGenre2", "userCity",
                       "authorid", "userRealplayCount", "videoRealPlayCount", "videoDuration", "videorealplayrate"]
    SEQ_LEN = 50
    negsample = 3
    features = ["videoid", "mid", "mobileos",
                "videoGenre1", "videoGenre2", "userRatedVideo1", "userRatedVideo2", "userRatedVideo3", "userGenre1", "userGenre2", "userCity",
                "authorid", "userRealplayCount", "videoRealPlayCount", "videoDuration", "videorealplayrate"]

    feature_max_idx = {}
    data["mid"].replace("unknown", "N000111111D", inplace=True)
    data = data[data["mid"] != "unknown"].copy()
    data = data[data["mid"] != "N000111111D"].copy()


    # 和上面函数的功能是一样的，见 deepMatch  DSSM
    def add_index_column(param_df, column_name):
        values = list(param_df[column_name].unique())
        value_index_dict = {value: idx for idx, value in enumerate(values)}
        if column_name == "mid":
            param_df["uidx"] = param_df[column_name].copy()
            param_df["mid"] = param_df[column_name].map(value_index_dict) + 1
            feature_max_idx["mid"] = param_df["mid"].max() + 1


    add_index_column(data, "mid")

    feature_max_idx["videoid"] = data["videoid"].max() + 1
    for idx, column_name in enumerate(features):

        lbe = LabelEncoder()

        if column_name == "videoGenre1" or column_name == "videoGenre2" or \
                column_name == "videoGenre3" or column_name == "userGenre1" or column_name == "userGenre2" or column_name == "userGenre3":
            data[column_name].fillna("社会", inplace=True)

        if column_name == "userCity":
            data[column_name].fillna("北京", inplace=True)

        if column_name == "mid":
            continue
            data["uidx"] = data[column_name].copy()
            data["mid"] = lbe.fit_transform(data[column_name])
            feature_max_idx["mid"] = data["mid"].max() + 1
        elif column_name == "videoid":   # 负采样生成的videoid，没有离散化
            continue
            data["vidx"] = data[column_name].copy()
            data["videoid"] = lbe.fit_transform(data[column_name])
            feature_max_idx["videoid"] = data["videoid"].max() + 1
        else:
            data[column_name] = lbe.fit_transform(data[column_name])
            feature_max_idx[column_name] = data[column_name].max() + 1

    user_profile = data[["uidx", "mid", "mobileos", "userRatedVideo1", "userRatedVideo2", "userRatedVideo3", "userGenre1", "userGenre2", "userCity", "userRealplayCount",
                         "videoGenre1", "videoGenre2", "authorid", "videoRealPlayCount", "videoDuration", "videorealplayrate"]].drop_duplicates('mid')

    user_mid_uid = data[["uidx", "mid"]].drop_duplicates('mid')

    user_mid_uid.rename(columns={'mid': 'userid'}, inplace=True)

    item_profile = data[
        ["videoid", "videoGenre1", "videoGenre2", "authorid", "videoRealPlayCount", "videoDuration", "videorealplayrate"]].drop_duplicates(
        'videoid')
    print("item size is:  ", len(item_profile))
    user_profile.set_index("mid", inplace=True)
    print(data)
    print("\n\n after group by mid videoid")
    # print(data)
    del data
    gc.collect()
    # 按序列采样，没有加负采样，会再训练时的batch里进行batch负采样
    # test_path = "/work/xielixun/test_user_video_play_test.csv"
    # train_path = "/work/xielixun/train_user_video_play.csv"
    # 负采样数据
    # test_path = "/work/xielixun/test_user_video_play_test_sample_negtive.csv"
    # train_path = "/work/xielixun/train_user_video_play_sample_negtive.csv"
    test_path = "/work/xielixun/sampleNegtive/test_set_0422.csv"
    train_path = "/work/xielixun/sampleNegtive/train_set_0422.csv"
    train_set, test_set = gen_data_set(train_path, test_path, user_mid_uid)
    embedding_dim = 32

    user_feature_columns = [SparseFeat('mid', feature_max_idx['mid'], embedding_dim),
                            SparseFeat("mobileos", feature_max_idx["mobileos"], embedding_dim),
                            SparseFeat("userRatedVideo1", feature_max_idx['userRatedVideo1'], embedding_dim),
                            SparseFeat("userRatedVideo2", feature_max_idx['userRatedVideo2'], embedding_dim),
                            SparseFeat("userRatedVideo3", feature_max_idx['userRatedVideo3'], embedding_dim),

                            SparseFeat("userGenre1", feature_max_idx['userGenre1'], embedding_dim),
                            SparseFeat("userGenre2", feature_max_idx['userGenre2'], embedding_dim),

                            SparseFeat("userCity", feature_max_idx['userCity'], embedding_dim),
                            SparseFeat("userRealplayCount", feature_max_idx['userRealplayCount'], embedding_dim),
                            VarLenSparseFeat(SparseFeat('hist_video_id', feature_max_idx['videoid'], embedding_dim,
                                                        embedding_name="videoid"), SEQ_LEN, 'mean', 'hist_len'),
                            ]

    item_feature_columns = [SparseFeat('videoid', feature_max_idx['videoid'], embedding_dim),
                            SparseFeat("videoGenre1", feature_max_idx["videoGenre1"], embedding_dim),
                            SparseFeat("videoGenre2", feature_max_idx["videoGenre2"], embedding_dim),
                            SparseFeat("authorid", feature_max_idx["authorid"], embedding_dim),
                            SparseFeat("videoRealPlayCount", feature_max_idx["videoRealPlayCount"], embedding_dim),
                            SparseFeat("videoDuration", feature_max_idx["videoDuration"], embedding_dim),
                            SparseFeat("videorealplayrate", feature_max_idx["videorealplayrate"], embedding_dim)]

    feature_names = get_feature_names(user_feature_columns + item_feature_columns)
    print("\n\nfeature_names is:   ")
    print(feature_names)

    # 因为下面这几行没有加，youtubeDNN导致了错误   tensorflow.python.framework.errors_impl.InvalidArgumentError:  assertion failed: [predictions must be <= 1] [Condition x <= y did not hold element-wise:] [x (functional_1/sampled_softmax_layer/ExpandDims:0) = ] [[0.944198132][1.15184534][1.00592339]...] [y (Cast_4/x:0) = ] [1]
    K.set_learning_phase(True)
    import tensorflow as tf

    if tf.__version__ >= '2.0.0':
        tf.compat.v1.disable_eager_execution()

    model = DSSM(user_feature_columns, item_feature_columns)
    logdir = os.path.join("log_callbacks_dssm")  # Tensorboard需要一个文件夹
    if not os.path.exists(logdir):
        os.mkdir(logdir)
    output_model_file = os.path.join(logdir,
                                     'dssm_model.h5')

    callbacks = [
        tf.keras.callbacks.TensorBoard(logdir),
        tf.keras.callbacks.ModelCheckpoint(output_model_file,
                                           save_best_only=True),
        tf.keras.callbacks.EarlyStopping(patience=5, min_delta=1e-5),
    ]

    METRICS = [
        tf.keras.metrics.TruePositives(name='tp'),
        tf.keras.metrics.FalsePositives(name='fp'),
        tf.keras.metrics.TrueNegatives(name='tn'),
        tf.keras.metrics.FalseNegatives(name='fn'),
        tf.keras.metrics.BinaryAccuracy(name='accuracy'),
        tf.keras.metrics.Precision(name='precision'),
        tf.keras.metrics.Recall(name='recall'),
        tf.keras.metrics.AUC(name='auc'),
        tf.keras.metrics.AUC(name='auc-ROC', curve='ROC'),
        tf.keras.metrics.AUC(name='auc-PRC', curve='PR')
    ]

    # compile the model, set loss function, optimizer and evaluation metrics
    model.compile(
        loss='binary_crossentropy',
        optimizer='adam',
        metrics=METRICS
    )
    model.fit_generator(generate_arrays_from_train(train_set, user_profile, SEQ_LEN),
                        steps_per_epoch=len(train_set) // batch_size,
                        epochs=2, max_queue_size=10, workers=1,
                        callbacks=callbacks, verbose=1, use_multiprocessing=False)

    model.save("./tensorflow_DSSM-027-tzld-1.h5")

    all_item_model_input = {"videoid": item_profile['videoid'].values,
                            "videoGenre1": item_profile['videoGenre1'].values,
                            "videoGenre2": item_profile['videoGenre2'].values,
                            "authorid": item_profile['authorid'].values,
                            "videoRealPlayCount": item_profile['videoRealPlayCount'].values,
                            "videoDuration": item_profile['videoDuration'].values,
                            "videorealplayrate": item_profile['videorealplayrate'].values}

    user_embedding_model = Model(inputs=model.user_input, outputs=model.user_embedding)
    item_embedding_model = Model(inputs=model.item_input, outputs=model.item_embedding)

    user_embs = user_embedding_model.predict_generator(generate_arrays_from_test(test_set, user_profile, SEQ_LEN),
                                                       steps=len(test_set) // batch_size,
                                                       max_queue_size=1, workers=1, callbacks=callbacks,
                                                       verbose=1, use_multiprocessing=False)

    test_model_input_emb2, test_model_label2 = gen_model_input_user_emb(test_set, user_profile, SEQ_LEN)
    user_embs2 = user_embedding_model.predict(test_model_input_emb2, batch_size=2 ** 12)

    item_embs = item_embedding_model.predict(all_item_model_input, batch_size=2 ** 12)
    # test_set里带有mid的原始值，可以取出作为生成embedding的key mid

    user_embs = list(map(str, user_embs2))
    user_res = {"mid": test_model_input_emb2["userid_org"], "emb": user_embs}
    user_embs_res = pd.DataFrame.from_dict(user_res)
    output_user = "./tensorflow_user_embedding-dssm-tzld-210423.csv"
    user_embs_res.to_csv(output_user, index=False)

    item_embs = list(map(str, item_embs))
    item_res = {"videoid": all_item_model_input["videoid"], "emb": item_embs}
    item_embs_res = pd.DataFrame.from_dict(item_res)
    output_item = "./tensorflow_video_embedding-dssm-tzld-210423.csv"
    item_embs_res.to_csv(output_item, index=False)

    print("************  21-04-23  train dssm cost time : " + str(time.time() - begin_time))