<jupyter_start><jupyter_text>Build Hybrid Recommendation SystemThe objective of this notebook is to create news_recommend_dataset and build a hybrid recommedation system to recommend news for Austrian news website [Kurier.at](https://kurier.at/). The hybrid recommedation model takes content-based features and latent factors as input, and predict the next news aritcle the user may be interested in. First, we combine preprocess dataset with user and item latent factors to create news_recommend_dataset. Next, we apply wide and deep architecture for our hybrid recommendation model. The deep network takes dense embedding features, and the wide network takes sparse features. Finally, create the model with version on ai-platform to perform batch prediction and online prediction.---The implementation is related to the following paper:- [Wide & Deep Learning for Recommender Systems](https://arxiv.org/abs/1606.07792)  1. import libraries<jupyter_code># import libraries
import os
import glob
import pandas as pd
import json
import requests

from google.cloud import bigquery
from oauth2client.client import GoogleCredentials
# set constants
PROJECT = "hybrid-recsys-gcp"
BUCKET = "hybrid-recsys-gcp-bucket"
REGION = 'us-central1'
DATASET = 'news_recommend_dataset'
TABLE = "news_recommend"
MODEL="hybrid_recsys_trained_model"
VERSION="v1"

os.environ["PROJECT"] = PROJECT
os.environ["BUCKET"] = BUCKET
os.environ["REGION"] = REGION
os.environ["DATASET"] = DATASET
os.environ["TABLE"] = TABLE
os.environ["MODEL"] = MODEL
os.environ["VERSION"] = VERSION<jupyter_output><empty_output><jupyter_text>2. create news_recommend_datasetCombine preprocess dataset with user and item latent factors to create "news_recommend_train.csv" and "news_recommend_test.csv".<jupyter_code>def store_query_result_to_table(query, project_id, dataset_id, table_id, bucket_id, schema, mode):
    """ Execute query, store result in bigquery table, and save result to bucket in csv file.
    
    Args:
        query (str): The query to be executed in bigquery.
        project_id (str): The ID of your project.
        dataset_id (str): The name of the dataset.
        table_id (str): The name for the table.
        bucket_id (str): Bucket to store the csv file of query result.
        schema (list:bigquery.SchemaField): Schema of the query result.
        mode (str): Train or test mode.

    Returns:
        None
    """
    client = bigquery.Client(project=project_id)
    
    table = bigquery.Table("{}.{}.{}".format(project_id, dataset_id, table_id + "_" + mode), schema=schema)
    table = client.create_table(table)
    table_ref = client.dataset(dataset_id).table(table_id + "_" + mode)
    
    job_config = bigquery.QueryJobConfig()
    job_config.destination = table_ref
    job_config.write_disposition = bigquery.WriteDisposition.WRITE_EMPTY
        
    query_job = client.query(query, job_config=job_config)
    result = query_job.result()
    
    destination_uri = "gs://{}/{}/{}_{}.csv".format(bucket_id, dataset_id, table_id, mode)
    extract_job = client.extract_table(table_ref, destination_uri)
    extract_job.result()
# specify train and test query for hybrid dataset
hybrid_train_query = """
SELECT *
FROM {}.preprocess_train

INNER JOIN {}.user_latent
USING (user_id)

INNER JOIN {}.item_latent
USING (item_id)
""".format(DATASET, DATASET, DATASET)

hybrid_test_query = """
SELECT *
FROM {}.preprocess_test

INNER JOIN {}.user_latent
USING (user_id)

INNER JOIN {}.item_latent
USING (item_id)
""".format(DATASET, DATASET, DATASET)

# specify schema
latent_num = 10

schema = [
    bigquery.SchemaField("user_id", "STRING", mode="REQUIRED"),
    bigquery.SchemaField("item_id", "STRING", mode="REQUIRED"),
    bigquery.SchemaField("title", "STRING", mode="REQUIRED"),
    bigquery.SchemaField("author", "STRING", mode="REQUIRED"),
    bigquery.SchemaField("category", "STRING", mode="REQUIRED"),
    bigquery.SchemaField("device_brand", "STRING", mode="REQUIRED"),
    bigquery.SchemaField("article_year", "STRING", mode="REQUIRED"),
    bigquery.SchemaField("article_month", "STRING", mode="REQUIRED"),
    bigquery.SchemaField("rating", "FLOAT", mode="REQUIRED"),
    bigquery.SchemaField("next_item_id", "STRING", mode="REQUIRED"),
    bigquery.SchemaField("fold", "INTEGER", mode="REQUIRED")
    ] + \
    [bigquery.SchemaField("user_latent_{}".format(i), "FLOAT", mode="REQUIRED") for i in range(2 * latent_num)] + \
    [bigquery.SchemaField("item_latent_{}".format(i), "FLOAT", mode="REQUIRED") for i in range(2 * latent_num)]

# create table
store_query_result_to_table(hybrid_train_query, PROJECT, DATASET, TABLE, BUCKET, schema, "train")
store_query_result_to_table(hybrid_test_query, PROJECT, DATASET, TABLE, BUCKET, schema, "test")<jupyter_output><empty_output><jupyter_text>3. view hybrid dataset<jupyter_code>!gsutil cp gs://{BUCKET}/{DATASET}/{TABLE}_train.csv ./{DATASET}/{TABLE}_train.csv
!gsutil cp gs://{BUCKET}/{DATASET}/{TABLE}_test.csv ./{DATASET}/{TABLE}_test.csv
train_df = pd.read_csv("./{}/{}_train.csv".format(DATASET, TABLE))
test_df = pd.read_csv("./{}/{}_test.csv".format(DATASET, TABLE))
train_df.head()
test_df.head()<jupyter_output><empty_output><jupyter_text>4. create hybrid_recsys packageThis is the package for the hybrid recommendation model. The trainer/task.py defines the input arguments for training. In trainer/model.py, the "create_dataset" function creates tf.dataset for input features. The "Hybrid_Recsys_Model" class defines the tf.keras.Model to take feature inputs and predict the next item the viewer may be interested in. And the "train_and_export_model" function defines the custom training loop for training the model. The Hybrid_Recsys_Model use the following architecture:<jupyter_code>%%bash
mkdir -p hybrid_recsys/trainer
touch hybrid_recsys/trainer/__init__.py
%%writefile hybrid_recsys/setup.py

from setuptools import setup, find_packages

REQUIRED_PACKAGES = ['tensorflow-hub==0.8.0']

setup(name='trainer',
    version='0.1',
    packages=find_packages(),
    install_requires=REQUIRED_PACKAGES,
    include_package_data=True,
    description='Setup dependencies for trainer package.'
)
%%writefile hybrid_recsys/trainer/task.py

import argparse
import tensorflow as tf
from trainer import model

if __name__ == "__main__":
    parser = argparse.ArgumentParser()
    parser.add_argument(
        "--job-dir",
        help="job dir to store training outputs and other data",
        required=True
    )
    
    parser.add_argument(
        "--train_data_path",
        help="path to import train data",
        required=True
    )
    
    parser.add_argument(
        "--test_data_path",
        help="path to import test data",
        required=True
    )
    
    parser.add_argument(
        "--output_dir",
        help="output dir to export checkpoints or trained model",
        required=True
    )
    
    parser.add_argument(
        "--batch_size",
        help="batch size for training",
        type=int,
        default=2048
    )
    
    parser.add_argument(
        "--epochs",
        help="number of epochs for training",
        type=int,
        default=1
    )
    
    parser.add_argument(
        "--latent_num",
        help="number of latent factors for gmf and mlp",
        type=int,
        default=10
    )

    parser.add_argument(
        "--item_id_path",
        help="path to import item_id_list.txt",
        required=True
    )
    
    parser.add_argument(
        "--author_path",
        help="path to import author_list.txt",
        required=True
    )
    
    parser.add_argument(
        "--category_path",
        help="path to import category_list.txt",
        required=True
    )
    
    parser.add_argument(
        "--device_brand_path",
        help="path to import device_brand_list.txt",
        required=True
    )
    parser.add_argument(
        "--article_year_path",
        help="path to import article_year_list.txt",
        required=True
    )
    
    parser.add_argument(
        "--article_month_path",
        help="path to import article_month_list.txt",
        required=True
    )
    
    parser.add_argument(
        "--save_tb_log_to_bucket",
        help="set to save tensorboard logs in bucket",
        default=False,
        action="store_true"
    )
    
    parser.add_argument(
        "--bucket_tb_log_path",
        help="path to store tensorboard in gcp bucket",
        default="gs://hybrid-recsys-gcp-bucket/tensorboard_log/ "
    )
    
    
    args = parser.parse_args()
    args = args.__dict__

    model.train_and_export_model(args)
%%writefile hybrid_recsys/trainer/model.py

import os
import pandas as pd
import tensorflow as tf
import tensorflow_hub as hub
import shutil
import datetime

# create dataset function
def create_dataset(path, column_name, label_name, defaults, batch_size, shuffle):
    """ Create tf.dataset from csv file.
    
    Args:
        path (str): Path to the csv file.
        column_names (list:str): List of string to specify which columns to use in dataset (including label).
        label_name (str): Column name for the label.
        defaults (list:str): List of string to set default values for columns.
        batch_size (str): Batchsize of the dataset.
        shuffle (bool): True for shuffling dataset and False otherwise.

    Returns:
        (tf.dataset): dataset used for training or testing
    """
    dataset = tf.data.experimental.make_csv_dataset(
        file_pattern = path,
        select_columns=column_name,
        label_name=label_name,
        column_defaults = defaults,
        batch_size=batch_size,
        num_epochs=1,
        shuffle=shuffle
    )
    return dataset

# model class
class Hybrid_Recsys_Model(tf.keras.Model):
    """ The Hybrid_Recsys_Model class. For training, the model takes input features and predict the probability for index of next item id. 
    For serving, the model takes input features and predict the next item id directly.

    Attributes:
        user_id_hash (tf.feature_column.categorical_column_with_hash_bucket): hash bucket column for user_id.
        item_id_hash (tf.feature_column.categorical_column_with_hash_bucket): hash bucket column for item_id.
        author_hash (tf.feature_column.categorical_column_with_hash_bucket): hash bucket column for author.
        device_brand_hash (tf.feature_column.categorical_column_with_hash_bucket): hash bucket column for device_brand.
        
        user_id_embed (tf.feature_column.embedding_column): embedding column for user_id.
        item_id_embed (tf.feature_column.embedding_column): embedding column for item_id.
        author_embed (tf.feature_column.embedding_column): embedding column for author.
        device_brand_embed (tf.feature_column.embedding_column): embedding column for device_brand.
        
        author_vocab (tf.feature_column.categorical_column_with_vocabulary_list): vocabulary list column for author.
        article_year_vocab (tf.feature_column.categorical_column_with_vocabulary_list): vocabulary list column for article_year.
        article_month_vocab (tf.feature_column.categorical_column_with_vocabulary_list): vocabulary list column for article_month.
        category_vocab (tf.feature_column.categorical_column_with_vocabulary_list): vocabulary list column for category.
        device_brand_vocab (tf.feature_column.categorical_column_with_vocabulary_list): vocabulary list column for device_brand.
        
        author_indicator (tf.feature_column.indicator_column): indicator colunm for author.
        cross_date_indicator (tf.feature_column.indicator_column): indicator colunm for crossing article_year and article_month.
        category_indicator (tf.feature_column.indicator_column): indicator colunm for category.
        device_brand_indicator (tf.feature_column.indicator_column): indicator colunm for device_brand.
        
        cross_date (tf.feature_column.crossed_column): crossed column for crossing rticle_year and article_month.
        
        u_latent_numeric (list:tf.feature_column.numeric_column): list of numeric columns for user latent factors.
        i_latent_numeric (list:tf.feature_column.numeric_column): list of numeric columns for item latent factors
        
        feature_columns_d (list:tf.feature_column): list of feature columns for deep features.
        feature_columns_w (list:tf.feature_column): list of feature columns for wide features.
        
        feature_layer_d (tf.keras.layers.DenseFeatures): dense layer for deep feature column.
        feature_layer_d (tf.keras.layers.DenseFeatures): dense layer for wide feature column.
        
        text_embed_layer (hub.KerasLayer): tenseorflow hub layer (NNLM) for text embedding
        
        dense_1 (tf.keras.layers.Dense): first dense layer for deep network
        dense_2 (tf.keras.layers.Dense): second dense layer for deep network
        dense_3 (tf.keras.layers.Dense): third dense layer for deep network
        dense_4 (tf.keras.layers.Dense): output layer which takes wide and deep networks and predict index for next item.
        
        item_id_table (tf.lookup.StaticHashTable): table for converting index to ids
    """
    def __init__(self, item_id_path, author_path, category_path, device_brand_path, article_year_path, article_month_path, latent_num):
        """ init method for Hybrid_Recsys_Model class
        
        Args:
            item_id_path (str): Path to txt file containing unique item ids.
            author_path (str): Path to txt file containing unique authors.
            category_path (str): Path to txt file containing unique categories.
            device_brand_path (str): Path to txt file containing unique device_brands.
            article_year_path (str): Path to txt file containing unique article_years.
            article_month_path (str): Path to txt file containing unique article_months.
            latent_num (int): Number of laten factors (gmf and mlp stream each) for representing user ids and item ids.
            
        Returns:
            None
        """
        super(Hybrid_Recsys_Model, self).__init__()
        # user_id embed
        self.user_id_hash = tf.feature_column.categorical_column_with_hash_bucket('user_id', 100)
        self.user_id_embed = tf.feature_column.embedding_column(categorical_column=self.user_id_hash, dimension=5)

        # item_id embed
        self.item_id_hash = tf.feature_column.categorical_column_with_hash_bucket('item_id', 200)
        self.item_id_embed = tf.feature_column.embedding_column(categorical_column=self.item_id_hash, dimension=20)

        # author embed
        self.author_hash = tf.feature_column.categorical_column_with_hash_bucket('author', 20)
        self.author_embed = tf.feature_column.embedding_column(categorical_column=self.author_hash, dimension=10)
        
        # author indicator
        self.author_vocab = tf.feature_column.categorical_column_with_vocabulary_list('author', self.get_list(author_path))
        self.author_indicator = tf.feature_column.indicator_column(self.author_vocab)

        # cross article year, month
        self.article_year_vocab = tf.feature_column.categorical_column_with_vocabulary_list('article_year', self.get_list(article_year_path))
        self.article_month_vocab = tf.feature_column.categorical_column_with_vocabulary_list('article_month', self.get_list(article_month_path))
        
        self.cross_date = tf.feature_column.crossed_column([self.article_year_vocab, self.article_month_vocab], \
                                                           self.get_size(article_year_path) * self.get_size(article_month_path))
        self.cross_date_indicator = tf.feature_column.indicator_column(self.cross_date)

        # category indicator
        self.category_vocab = tf.feature_column.categorical_column_with_vocabulary_list('category', self.get_list(category_path))
        self.category_indicator = tf.feature_column.indicator_column(self.category_vocab)

        # device_brand embed
        self.device_brand_hash= tf.feature_column.categorical_column_with_hash_bucket('device_brand', 10)
        self.device_brand_embed = tf.feature_column.embedding_column(categorical_column=self.device_brand_hash, dimension=5)

        # device_brand indicator
        self.device_brand_vocab = tf.feature_column.categorical_column_with_vocabulary_list('device_brand', self.get_list(device_brand_path))
        self.device_brand_indicator = tf.feature_column.indicator_column(self.device_brand_vocab)

        # user item latent factors
        self.u_latent_numeric = [tf.feature_column.numeric_column(key="user_latent_" + str(i)) for i in range(2 * latent_num)]
        self.i_latent_numeric =  [tf.feature_column.numeric_column(key="item_latent_" + str(i)) for i in range(2 * latent_num)]

        # combine feature columns to feature layer
        self.feature_columns_d =  [self.user_id_embed, self.item_id_embed, self.author_embed, self.device_brand_embed] \
                                    + self.u_latent_numeric + self.i_latent_numeric
        self.feature_layer_d = tf.keras.layers.DenseFeatures(self.feature_columns_d)
        
        self.feature_columns_w = [self.author_indicator, self.cross_date_indicator, self.category_indicator, self.device_brand_indicator]
        self.feature_layer_w = tf.keras.layers.DenseFeatures(self.feature_columns_w)

        # title tf_hub nnlm embedding
        self.text_embed_layer = hub.KerasLayer("https://tfhub.dev/google/nnlm-de-dim50-with-normalization/2", dtype=tf.string)
       
        # dense
        self.dense_1 = tf.keras.layers.Dense(200, activation='relu')
        self.dense_2 = tf.keras.layers.Dense(100, activation='relu')
        self.dense_3 = tf.keras.layers.Dense(50, activation='relu')
        self.dense_4 = tf.keras.layers.Dense(self.get_size(item_id_path) + 1, activation='softmax')

        # item_id lookup table
        self.item_id_table = self.create_item_id_table(item_id_path)
    
    @tf.function
    def call(self, inputs):
        """The call method for NeuMF class.

        Args:
            inputs (OrderedDict:tf.Tensor): OrderedDict of input feature tensor.
        
        Returns:
            output (tf.Tensor): The predicted probability for index of next item id.
        """
        # wide, deep feature columns
        feature_cols_d = self.feature_layer_d(inputs)
        feature_cols_w = self.feature_layer_w(inputs)
        
        # title embedding
        title = inputs['title']
        title_embed = self.text_embed_layer(title)
        
        # deep network
        concat_1 = tf.concat([feature_cols_d, title_embed], axis=1)
        dense_1_out = self.dense_1(concat_1)
        dense_2_out = self.dense_2(dense_1_out)
        dense_3_out = self.dense_3(dense_2_out)
        
        # combine wide, and deep layers
        concat_2 = tf.concat([dense_3_out, feature_cols_w], axis=1)

        output = self.dense_4(concat_2)
        return output
    
    
    latent_num = 10
    key = ["user_id", "item_id", "title", "author", "category", "device_brand", "article_year", "article_month"] + \
            ["user_latent_{}".format(i) for i in range(2*latent_num)] + ["item_latent_{}".format(i) for i in range(2*latent_num)]
    value = [tf.TensorSpec([None], dtype=tf.string, name="user_id"),
                tf.TensorSpec([None], dtype=tf.string, name="item_id"), \
                tf.TensorSpec([None], dtype=tf.string, name="title"), \
                tf.TensorSpec([None], dtype=tf.string, name="author"), \
                tf.TensorSpec([None], dtype=tf.string, name="category"), \
                tf.TensorSpec([None], dtype=tf.string, name="device_brand"), \
                tf.TensorSpec([None], dtype=tf.string, name="article_year"), \
                tf.TensorSpec([None], dtype=tf.string, name="article_month")] + \
            [tf.TensorSpec([None], dtype=tf.float32, name="user_latent_{}".format(i)) for i in range(2*latent_num)] + \
            [tf.TensorSpec([None], dtype=tf.float32, name="item_latent_{}".format(i)) for i in range(2*latent_num)]
    signature_dict = dict(zip(key, value))
    @tf.function(input_signature=[signature_dict])
    def my_serve(self, x):
        """The serving method for Hybrid class.

        Args:
            inputs (OrderedDict:tf.Tensor): OrderedDict of input feature tensor.
        
        Returns:
            output (tf.Tensor): The predicted next item id.
        """
        pred = self.__call__(x)
        values, indices = tf.math.top_k(pred, k=10)
        item_ids = self.item_id_table.lookup(indices)
        return {"top_k_item_id": item_ids}
    
    def get_size(self, file_path):
        """Returns total number of lines in the txt file.

        Args:
            file_path (str): Path to txt file.
        
        Returns:
            size (int): total number of lines in the txt file.
        """
        id_tensors = tf.strings.split(tf.io.read_file(file_path), '\n')
        return id_tensors.shape[0]
    
    def get_list(self, file_path):
        """Extract content from txt file and store into list. Each line in txt file corresponds to an elemnt in list.

        Args:
            file_path (str): Path to txt file.
        
        Returns:
            id_list (int): list of values from txt file.
        """
        id_tensors = tf.strings.split(tf.io.read_file(file_path), '\n')
        id_list = [tf.compat.as_str_any(x) for x in id_tensors.numpy()]
        return id_list
    
    def create_item_id_table(self, file_path):
        """ create lookup table to translate item index to item id.
        
        Args:
            file_path (str): Path to txt file containing item ids.
            
        Returns:
            (tf.lookup.StaticVocabularyTable): The lookup table.
        """
        values = tf.strings.split(tf.io.read_file(file_path), '\n')
        keys = tf.range(values.shape[0], dtype=tf.int32)
        initializer = tf.lookup.KeyValueTensorInitializer(keys=keys, values=values, key_dtype=tf.int32, value_dtype=tf.string)
        table = tf.lookup.StaticHashTable(initializer, default_value='unknown')
        return table
        
        
    
def train_and_export_model(args):
    """ Train the Hybrid_Recsys_Model and export model to bucket.

    Args:
        args (dict): dict of arguments from task.py

    Returns:
        None
    """
    # create dataset
    feature_col = ['user_id', 'item_id', 'title', 'author', 'category', 'device_brand', 'article_year', 'article_month', 'next_item_id']
    u_latent_col = ["user_latent_{}".format(i) for i in range(2 * args["latent_num"])]
    i_latent_col = ["item_latent_{}".format(i) for i in range(2 * args["latent_num"])]
    column_name = feature_col + u_latent_col + i_latent_col
    
    label_name = 'next_item_id'
    defaults = ['unknown'] * len(feature_col) + [0.0] * (len(u_latent_col) + len(i_latent_col))
    batch_size = args["batch_size"]
    train_path = args["train_data_path"]
    test_path = args["test_data_path"]
    
    train_dataset = create_dataset(train_path, column_name, label_name, defaults, batch_size, True)
    test_dataset = create_dataset(test_path, column_name, label_name, defaults, batch_size, False)
    
    # create model
    model = Hybrid_Recsys_Model(args["item_id_path"], args["author_path"], args["category_path"], args["device_brand_path"], \
                                args["article_year_path"], args["article_month_path"], args["latent_num"])
    
    # loopup table (convvert id to index)
    item_index_initializer = tf.lookup.TextFileInitializer(args["item_id_path"], key_dtype=tf.string, key_index=tf.lookup.TextFileIndex.WHOLE_LINE, \
                            value_dtype=tf.int64, value_index=tf.lookup.TextFileIndex.LINE_NUMBER, delimiter="\n")
    item_index_table = tf.lookup.StaticVocabularyTable(item_index_initializer, num_oov_buckets=1)
    
    
    # loss function and optimizers
    loss_object = tf.keras.losses.SparseCategoricalCrossentropy()
    optimizer = tf.keras.optimizers.Adam(learning_rate = 0.001)
    
    # loss metrics
    train_loss = tf.keras.metrics.Mean(name='train_loss')
    train_accuracy = tf.keras.metrics.SparseCategoricalAccuracy(name='train_accuracy')
    train_top_10_accuracy = tf.keras.metrics.SparseTopKCategoricalAccuracy(k=10, name='train_top_10_accuracy')

    test_loss = tf.keras.metrics.Mean(name='test_loss')
    test_accuracy = tf.keras.metrics.SparseCategoricalAccuracy(name='test_accuracy')
    test_top_10_accuracy = tf.keras.metrics.SparseTopKCategoricalAccuracy(k=10, name='test_top_10_accuracy')
    
    # tensorboard
    current_time = datetime.datetime.now().strftime("%Y%m%d-%H%M%S")
    train_log_dir = './logs/gradient_tape/' + current_time + '/train'
    test_log_dir = './logs/gradient_tape/' + current_time + '/test'
    train_summary_writer = tf.summary.create_file_writer(train_log_dir)
    test_summary_writer = tf.summary.create_file_writer(test_log_dir)
    
    @tf.function
    def train_step(features, labels):
        """ Concrete function for train setp and update train metircs

        Args:
            features (OrderedDict:tf.Tensor): OrderedDict of tensor containing input  features.
            labels (tf.Tensor): labels indicating the next_item id

        Returns:
            None
        """
        with tf.GradientTape() as tape:
            predictions = model(features, training=True)
            label_indicies = item_index_table.lookup(labels)
            loss = loss_object(label_indicies, predictions)
        gradients = tape.gradient(loss, model.trainable_variables)
        optimizer.apply_gradients(zip(gradients, model.trainable_variables))

        train_loss(loss)
        train_accuracy(label_indicies, predictions)
        train_top_10_accuracy(label_indicies, predictions)
    
    @tf.function
    def test_step(features, labels):
        """ Concrete function for test setp and update test metircs

        Args:
            features (OrderedDict:tf.Tensor): OrderedDict of tensor containing input features.
            labels (tf.Tensor): labels indicating the next_item id

        Returns:
            None
        """
        predictions = model(features, training=False)
        label_indicies = item_index_table.lookup(labels)
        loss = loss_object(label_indicies, predictions)

        test_loss(loss)
        test_accuracy(label_indicies, predictions)
        test_top_10_accuracy(label_indicies, predictions)
    
    # custom train loop
    EPOCHS = args["epochs"]

    for epoch in range(EPOCHS):
        train_loss.reset_states()
        train_accuracy.reset_states()
        train_top_10_accuracy.reset_states()

        test_loss.reset_states()
        test_accuracy.reset_states()
        test_top_10_accuracy.reset_states()

        for features, labels in train_dataset:
            train_step(features, labels)
        with train_summary_writer.as_default():
            tf.summary.scalar('loss', train_loss.result(), step=epoch)
            tf.summary.scalar('accuracy', train_accuracy.result(), step=epoch)
            tf.summary.scalar('top_10_accuracy', train_top_10_accuracy.result(), step=epoch)

        for features, labels in test_dataset:
            test_step(features, labels)
        with test_summary_writer.as_default():
            tf.summary.scalar('loss', test_loss.result(), step=epoch)
            tf.summary.scalar('accuracy', test_accuracy.result(), step=epoch)
            tf.summary.scalar('top_10_accuracy', test_top_10_accuracy.result(), step=epoch)

        template = 'Epoch {:d}, train[loss: {:.6f}, acc: {:.6f}, top_10_acc: {:.6f}], Test[loss: {:.6f}, acc: {:.6f}, top_10_acc: {:.6f}]'

        print(template.format(epoch + 1,
                              train_loss.result(),
                              train_accuracy.result() * 100,
                              train_top_10_accuracy.result() * 100,

                              test_loss.result(),
                              test_accuracy.result() * 100,
                              test_top_10_accuracy.result() * 100,
                              ))
    
    # exprot tensorboard log
    if args["save_tb_log_to_bucket"]:
        script = "gsutil cp -r ./logs {}".format(args["bucket_tb_log_path"])
        os.system(script)
    
    # export model
    EXPORT_PATH = os.path.join(args["output_dir"], datetime.datetime.now().strftime("%Y%m%d%H%M%S"))
    tf.saved_model.save(obj=model, export_dir=EXPORT_PATH, signatures={'serving_default': model.my_serve})<jupyter_output>Writing hybrid_recsys/trainer/model.py<jupyter_text>5. train model locallyRun package as a python module in local environment.<jupyter_code>%%bash

JOBDIR=./${MODEL}
OUTDIR=./${MODEL}

rm -rf ${JOBDIR}
export PYTHONPATH=${PYTHONPATH}:${PWD}/hybrid_recsys

python -m trainer.task \
    --job-dir=${JOBDIR} \
    --train_data_path=gs://${BUCKET}/${DATASET}/${TABLE}_train.csv \
    --test_data_path=gs://${BUCKET}/${DATASET}/${TABLE}_test.csv \
    --output_dir=${OUTDIR} \
    --batch_size=2048 \
    --epochs=40 \
    --latent_num=10 \
    --item_id_path=gs://${BUCKET}/${DATASET}/item_id_list.txt \
    --author_path=gs://${BUCKET}/${DATASET}/author_list.txt \
    --category_path=gs://${BUCKET}/${DATASET}/category_list.txt \
    --device_brand_path=gs://${BUCKET}/${DATASET}/device_brand_list.txt \
    --article_year_path=gs://${BUCKET}/${DATASET}/article_year_list.txt \
    --article_month_path=gs://${BUCKET}/${DATASET}/article_month_list.txt<jupyter_output>Epoch 1, train[loss: 6.038098, acc: 1.906627, top_10_acc: 17.033777], Test[loss: 5.349120, acc: 2.306068, top_10_acc: 19.401793]
Epoch 2, train[loss: 5.288680, acc: 2.978423, top_10_acc: 20.357052], Test[loss: 5.276124, acc: 2.916833, top_10_acc: 22.016096]
Epoch 3, train[loss: 5.221677, acc: 3.328410, top_10_acc: 22.089649], Test[loss: 5.207056, acc: 3.299275, top_10_acc: 22.935099]
Epoch 4, train[loss: 5.141301, acc: 3.820961, top_10_acc: 23.950039], Test[loss: 5.102207, acc: 4.315315, top_10_acc: 25.229750]
Epoch 5, train[loss: 5.006400, acc: 4.811200, top_10_acc: 27.856409], Test[loss: 5.009177, acc: 5.182944, top_10_acc: 28.551859]
Epoch 6, train[loss: 4.870391, acc: 5.633830, top_10_acc: 31.545082], Test[loss: 4.858266, acc: 5.519722, top_10_acc: 32.553230]
Epoch 7, train[loss: 4.736486, acc: 6.245826, top_10_acc: 34.408554], Test[loss: 4.763878, acc: 6.467264, top_10_acc: 34.648098]
Epoch 8, train[loss: 4.646415, acc: 6.771770, top_10_acc: 36.219494], Test[loss: 4.702007, acc: 6[...]<jupyter_text>6. train model on cloudSubmit a training job in gcloud ai-platform to train the package.<jupyter_code>%%bash

JOBDIR=gs://${BUCKET}/${MODEL}
OUTDIR=gs://${BUCKET}/${MODEL}
JOBID=hybrid_recsys_train_job_$(date -u +%y%m%d_%H%M%S)

gcloud ai-platform jobs submit training ${JOBID} \
    --region=${REGION} \
    --module-name=trainer.task \
    --package-path=$(pwd)/hybrid_recsys/trainer \
    --staging-bucket=gs://${BUCKET} \
    --scale-tier=CUSTOM \
    --master-machine-type=n1-highcpu-16 \
    --runtime-version=2.1 \
    --python-version=3.7 \
    -- \
    --job-dir=${JOBDIR} \
    --train_data_path=gs://${BUCKET}/${DATASET}/${TABLE}_train.csv \
    --test_data_path=gs://${BUCKET}/${DATASET}/${TABLE}_test.csv \
    --output_dir=${OUTDIR} \
    --batch_size=2048 \
    --epochs=40 \
    --latent_num=10 \
    --item_id_path=gs://${BUCKET}/${DATASET}/item_id_list.txt \
    --author_path=gs://${BUCKET}/${DATASET}/author_list.txt \
    --category_path=gs://${BUCKET}/${DATASET}/category_list.txt \
    --device_brand_path=gs://${BUCKET}/${DATASET}/device_brand_list.txt \
    --article_year_path=gs://${BUCKET}/${DATASET}/article_year_list.txt \
    --article_month_path=gs://${BUCKET}/${DATASET}/article_month_list.txt \
    --save_tb_log_to_bucket \
    --bucket_tb_log_path=gs://${BUCKET}/tensorboard_log<jupyter_output>jobId: hybrid_recsys_train_job_200816_180754
state: QUEUED<jupyter_text>The training log should look like the following. The final test result is loss: 4.471430, acc: 9.715167, top_10_acc: 42.097153. The top 10 accuracy is around 42.09%, which means our model has 42.09% chance to correctly predict the next news article the visitor would like to view if our hybrid recommendation model recommend 10 items. If randomly picking 10 items from total 2421 news articles, the top 10 accuracy would only be 0.4130%. Our model has 100 times better top 10 accuracy than random picking. In Coud shell, type ``` tensorboard --logdir=gs://hybrid-recsys-gcp-bucket/tensorboard_log --port=8080 ``` to launch tensorboard. Click the Web Preview button in Coud shell to open Tensorboard.  6. create serving model on gcloudCreate AI Platform Model and set model version for serving.<jupyter_code>%%bash

MODEL_PATH=$(gsutil ls gs://$BUCKET/$MODEL/ | tail -1)

echo $MODEL_PATH

gcloud ai-platform models create ${MODEL} --regions=us-central1

gcloud ai-platform versions create ${VERSION} \
    --model ${MODEL} \
    --staging-bucket gs://${BUCKET} \
    --origin ${MODEL_PATH} \
    --framework 'tensorflow' \
    --runtime-version 2.1 \
    --python-version 3.7<jupyter_output>gs://hybrid-recsys-gcp-bucket/hybrid_recsys_trained_model/20200816182739/<jupyter_text>7. batch prediction on gcloudSubmit a predition job in gcloud to perform batch prediction for "batch_pred_inputs.json". Batch prediction is optimized for handling large json file.<jupyter_code>BATCH_PRED_FOLDER="hybrid_recsys_batch_pred"
os.environ["BATCH_PRED_FOLDER"] = BATCH_PRED_FOLDER
def create_input_json(dataset_path, sample_num, inputs_json_name, label_name):
    """ Samples from dataset and reate input json file and label txt file for prediction.

    Args:
        dataset_path (str): Path to csv dataset in gcp bucket.
        sample_num (int): Number of samples to draw.
        inputs_json_name (str): Name of json file to store input for prediction.
        label_name (str): Name of txt file storing labels.

    Returns:
        None
    """
    script = "gsutil cp {} ./dataset.csv".format(dataset_path)
    os.system(script)

    df = pd.read_csv("./dataset.csv")
    df = df.sample(n=sample_num)
    df = df.astype({'user_id':'str', 'item_id':'str', 'next_item_id':'str', 'article_year':'str', 'article_month':'str'})
    df = df.drop(columns=['rating', 'fold'])
    label = df.pop('next_item_id')
    
    inputs_text = df.to_json(orient='records')[1:-1].replace('},{', '} \n {')
    with open(inputs_json_name, 'w') as f:
        f.write(inputs_text)
        
    label_text = '\n'.join(list(label))
    with open(label_name, 'w') as f:
        f.write(label_text)
    
    script = "rm ./dataset.csv"
    os.system(script)
dataset_path="gs://{}/{}/{}_test.csv".format(BUCKET, DATASET, TABLE)
inputs_json_name="batch_pred_inputs.json"
label_name="batch_pred_labels.txt"
sample_num = 2000

# create prediction inputs
create_input_json(dataset_path, sample_num, inputs_json_name, label_name)

# copy files to folder and bucket
!mkdir -p $BATCH_PRED_FOLDER
!mv ./batch_pred_inputs.json ./$BATCH_PRED_FOLDER
!mv ./batch_pred_labels.txt ./$BATCH_PRED_FOLDER
!gsutil cp -r ./$BATCH_PRED_FOLDER/* gs://$BUCKET/$BATCH_PRED_FOLDER/
# view first input sample
!head -1 ./hybrid_recsys_batch_pred/batch_pred_inputs.json
%%bash

INPUT=gs://${BUCKET}/${BATCH_PRED_FOLDER}/batch_pred_inputs.json
OUTPUT=gs://${BUCKET}/${BATCH_PRED_FOLDER}/output/
JOBID=hybrid_recsys_pred_job_$(date -u +%y%m%d_%H%M%S)

gsutil rm -rf $OUTPUT

gcloud ai-platform jobs submit prediction ${JOBID} \
  --data-format TEXT \
  --region ${REGION} \
  --input-paths ${INPUT} \
  --output-path ${OUTPUT} \
  --model ${MODEL} \
  --version ${VERSION} \
  --max-worker-count 8
def check_batch_pred_acc(folder_name, result_path, label_path):
    """ Copy prediction result and label to folder in current path. Calculate and print the accuracy 
    for batch prediction result

    Args:
        folder_name (str): Name of the folder to work.
        result_path (str): Path to prediction result file.
        label_path (str): Path to true label file.

    Returns:
        None
    """
    script2 = "mkdir -p ./{}/output/".format(folder_name)
    script1 = "rm -r ./{}/output/*".format(folder_name)
    script3 = "gsutil cp {} ./{}/output/".format(result_path, folder_name)
    os.system("{} & {} & {}".format(script1, script2, script3))
    
    filenames = sorted(glob.glob("./{}/output/prediction.results*".format(folder_name)))
    with open("./{}/output/batch_pred_result.txt".format(folder_name), 'w') as outputfile:
        for file in filenames:
            with open(file, 'r') as inputfile:
                for line in inputfile:
                    outputfile.write(line)
                    
    script = "gsutil cp {} ./{}/output/batch_pred_labels.txt".format(label_path, folder_name)
    os.system(script)
    
    match = 0
    count = 0
    with open("./{}/output/batch_pred_result.txt".format(folder_name)) as f1, open("./{}/output/batch_pred_labels.txt".format(folder_name)) as f2:
        for line1, line2 in zip(f1, f2):
            count += 1
            if line2.strip() in line1:
                match += 1
    
    print("Accuracy: {:.2f} %".format(100*match/count))
batch_pred_result_path="gs://{}/{}/output/prediction.results*".format(BUCKET, BATCH_PRED_FOLDER)
batch_pred_label_path="gs://{}/{}/batch_pred_labels.txt".format(BUCKET, BATCH_PRED_FOLDER)
check_batch_pred_acc(BATCH_PRED_FOLDER, batch_pred_result_path, batch_pred_label_path)<jupyter_output>Accuracy: 42.10 %<jupyter_text>The batch prediction result for 2000 samples is has top 10 accuracy around 42.10%, which is similar top 10 accuracy during training.  8. online prediction on gcloudUse gcloud to sends "online_pred_inputs.json" as json string, and the gcloud will parse and print the response message in terminal. The online prediction is optimized for minimizing latency.<jupyter_code>ONLINE_PRED_FOLDER="hybrid_recsys_online_pred"
os.environ["ONLINE_PRED_FOLDER"] = ONLINE_PRED_FOLDER
dataset_path="gs://{}/{}/{}_test.csv".format(BUCKET, DATASET, TABLE)
inputs_json_name="online_pred_inputs.json"
label_name="online_pred_labels.txt"
sample_num = 10

# create prediction inputs
create_input_json(dataset_path, sample_num, inputs_json_name, label_name)

# copy files to folder and bucket
!mkdir -p $ONLINE_PRED_FOLDER
!mv ./online_pred_inputs.json ./$ONLINE_PRED_FOLDER
!mv ./online_pred_labels.txt ./$ONLINE_PRED_FOLDER
!gsutil cp -r ./$ONLINE_PRED_FOLDER/* gs://$BUCKET/$ONLINE_PRED_FOLDER/
# view first input sample
!head -1 ./hybrid_recsys_online_pred/online_pred_inputs.json
%%bash

JSON_PATH=./${ONLINE_PRED_FOLDER}/online_pred_inputs.json

gcloud ai-platform predict \
  --model ${MODEL} \
  --version ${VERSION} \
  --json-instances ${JSON_PATH}<jupyter_output>TOP_K_ITEM_ID
[u'299816215', u'187077794', u'299826775', u'299913368', u'299410466', u'299853016', u'299907204', u'299848776', u'299826767', u'299925700']
[u'299865757', u'299910994', u'299826775', u'299410466', u'299913368', u'299836255', u'299816215', u'299917726', u'299925086', u'299931241']
[u'299935287', u'299953030', u'299943426', u'299957318', u'299931241', u'299949793', u'299935266', u'299930679', u'299826775', u'299949869']
[u'299949793', u'299935287', u'299941050', u'299930679', u'299925700', u'299826775', u'299939900', u'299934703', u'299922662', u'299936493']
[u'299914459', u'299809748', u'299937546', u'299837992', u'299907275', u'299903877', u'299942840', u'299804319', u'299907267', u'299814183']
[u'299816215', u'299866366', u'299836255', u'299824032', u'299826767', u'299836841', u'299837992', u'299800704', u'299910994', u'299833840']
[u'299865757', u'299959410', u'299836255', u'299910994', u'299410466', u'299931241', u'299933565', u'299826775', u'298972803', u'299917726'][...]<jupyter_text>9. prediction with REST APISend HTTP request to model API and receive HTTP response to get prediction.<jupyter_code>def create_input_data(dataset_path, sample_num):
    script = "gsutil cp {} ./dataset.csv".format(dataset_path)
    os.system(script)
    
    df = pd.read_csv("./dataset.csv")
    df = df.sample(n=sample_num)
    df = df.astype({'user_id':'str', 'item_id':'str', 'next_item_id':'str', 'article_year':'str', 'article_month':'str'})
    df = df.drop(columns=['rating', 'fold'])
    label = df.pop('next_item_id')
    
    script = "rm ./dataset.csv"
    os.system(script)
    
    return {'instances':df.to_dict('records')}, label
dataset_path="gs://{}/{}/{}_test.csv".format(BUCKET, DATASET, TABLE)
sample_num = 5

data3, label = create_input_data(dataset_path, sample_num)
# view first sample
data3['instances'][0]
# prepare aip
token = GoogleCredentials.get_application_default().get_access_token().access_token
api = 'https://ml.googleapis.com/v1/projects/{}/models/{}/versions/{}:predict'.format(PROJECT, MODEL, VERSION)
headers = {'Authorization': 'Bearer ' + token }

# make request
response = requests.post(api, json=data3, headers=headers)
print(response.content)<jupyter_output><empty_output>
{'fraction_non_alphanumeric': 0.11753808653260207, 'fraction_numerical': 0.03293113954905545, 'mean_word_length': 3.5073610195561415, 'pattern_counts': {'":': 2, '<': 35, '<?xml version=': 0, '>': 35, 'https://': 4, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 3, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'chain_length': '10', 'dir': 'jupyter-structured-clean-dedup', 'id': '400784', 'license': 'MIT', 'n_tokens_mistral': 14172, 'n_tokens_neox': 13290, 'n_words': 2849, 'path': 'hybrid_recsys_train.ipynb', 'repo_name': 'ChienTeLee/test2', 'size': '41025'}