Deep_match_rank_emb/match_recall/youtube_tzld_match.py

# encoding: utf-8
import pandas as pd
import os
import gc
from deepctr.feature_column import SparseFeat, VarLenSparseFeat, get_feature_names
# from preprocess_tzld210315 import gen_data_set, gen_model_input, gen_model_input_user_emb
from preprocess_tzld210322_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 *
from deepmatch.utils import sampledsoftmaxloss

count_train = 1
count_test = 1

#batch_size = 1024
batch_size = 6000

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
    # batch_size = 8
    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 是我们的训练集包括标签，每一行的第一个是我们的图片路径，后面的是我们的独热化后的标签

    # global count
    # batch_size = 8
    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
    # batch_size = 8
    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__":

    data = pd.read_csvdata = pd.read_csv("/work/xielixun/dwa_sum_graphembedding_user_action_feature_app_20210225.csv")
    print(data[0:5])
    sparse_features = ["videoid", "mid",
                       "videoGenre1", "userRatedVideo1", "userGenre1", "userCity",
                       "authorid", "userRealplayCount", "videoRealPlayCount", "videoDuration"]
    SEQ_LEN = 50
    #negsample = 3
    negsample = 0

    # 1.Label Encoding for sparse features,and process sequence features with `gen_date_set` and `gen_model_input`

    # features = ['user_id', 'movie_id', 'gender', 'age', 'occupation', 'zip']
    features = ["videoid", "mid",
                "videoGenre1", "userRatedVideo1", "userGenre1", "userCity",
                "authorid", "userRealplayCount", "videoRealPlayCount", "videoDuration"]

    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)
            feature_max_idx["mid"] = param_df["mid"].max() + 1


    add_index_column(data, "mid")

    feature_max_idx["videoid"] = data["videoid"].max() + 1

    # add_index_column(df_data, "mid")

    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]) + 1
            feature_max_idx[column_name] = data[column_name].max() + 1

    user_profile = data[["uidx", "mid", "userRatedVideo1", "userGenre1", "userCity", "userRealplayCount",
                         "videoGenre1", "authorid", "videoRealPlayCount", "videoDuration"]].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", "authorid", "videoRealPlayCount", "videoDuration"]].drop_duplicates(
        'videoid')
    print("item size is:  ", len(item_profile))
    user_profile.set_index("mid", inplace=True)
    del data
    gc.collect()
    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"
    train_set, test_set = gen_data_set(train_path, test_path, user_mid_uid)

    embedding_dim = 16
    user_feature_columns = [SparseFeat('mid', feature_max_idx['mid'], embedding_dim),
                            SparseFeat("userRatedVideo1", feature_max_idx['userRatedVideo1'], embedding_dim),
                            SparseFeat("userGenre1", feature_max_idx['userGenre1'], 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("authorid", feature_max_idx["authorid"], embedding_dim),
                            SparseFeat("videoRealPlayCount", feature_max_idx["videoRealPlayCount"], embedding_dim),
                            SparseFeat("videoDuration", feature_max_idx["videoDuration"], embedding_dim)]

    feature_names = get_feature_names(user_feature_columns + item_feature_columns)
    print("\n\nfeature_names is:   ")
    print(feature_names)
    # 因为下面这几行没有加，导致了错误   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 = YoutubeDNN(user_feature_columns, item_feature_columns, num_sampled=3, user_dnn_hidden_units=(64, embedding_dim), dnn_use_bn=True, output_activation='sigmoid')

    logdir = os.path.join("log_callbacks_youtube")  # Tensorboard需要一个文件夹
    if not os.path.exists(logdir):
        os.mkdir(logdir)
    output_model_file = os.path.join(logdir,
                                     'youtube_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')
    ]

    model.compile(
        loss=sampledsoftmaxloss,
        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=1, workers=1,
                        callbacks=callbacks, verbose=1, use_multiprocessing=False)

    model.save("./tensorflow_youtubeDNN-026-tzld-1.h5")

    all_item_model_input = {"videoid": item_profile['videoid'].values,
                            "videoGenre1": item_profile['videoGenre1'].values,
                            "authorid": item_profile['authorid'].values,
                            "videoRealPlayCount": item_profile['videoRealPlayCount'].values,
                            "videoDuration": item_profile['videoDuration'].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=10, workers=1, callbacks=callbacks,
                                                       verbose=1, use_multiprocessing=False)

    item_embs = item_embedding_model.predict(all_item_model_input, batch_size=2 ** 12)

    # 得到user embedding
    user_layer_model = tf.keras.models.Model(
        inputs=[model.user_input],
        # outputs=model.get_layer("user_embedding").output
        outputs=model.user_embedding
    )

    user_embeddings = []

    # 得到video embedding
    video_layer_model = tf.keras.models.Model(
        inputs=[model.item_input],
        # outputs=model.get_layer("item_embedding").output
        outputs=model.item_embedding
    )

    video_embeddings = []
    for index, row in item_profile.iterrows():
        # video_id = row["vidx"]
        video_id = row["videoid"]

        video_input = [
            np.reshape(row["videoid"], [1, 1]),
            np.reshape(row["videoGenre1"], [1, 1]),
            # np.reshape(row["videoGenre2"], [1, 1]),
            np.reshape(row["authorid"], [1, 1]),
            np.reshape(row["videoRealPlayCount"], [1, 1]),
            np.reshape(row["videoDuration"], [1, 1])
        ]
        video_embedding = video_layer_model(video_input)

        embedding_str = ",".join([str(x) for x in video_embedding.numpy().flatten()])
        video_embeddings.append([video_id, embedding_str])

    df_video_embedding = pd.DataFrame(video_embeddings, columns=["video_id", "video_embedding"])
    df_video_embedding.head()

    output = "./tensorflow_video_embedding-youtubeDNN-tzld-210322.csv"
    df_video_embedding.to_csv(output, index=False)