这篇文章算是对这一个月以来天池比赛《CIKM 2019 EComm AI：用户行为预测》的一个总结，我们队伍第一次打这种“海量数据”的比赛，最终成绩能保留在第一页就已经很满意了。

这次比赛的要求是根据用户历史行为预测未来行为，对于特定的用户， 为其推荐最多50个TA未来3天可能会点击的商品，也就是根据用户的历史交互行为、用户基本属性、物品基本属性，来预测测试集内的用户可能点击50个商品，也就是我们一般做推荐系统中，需要得到的那个推荐池。比赛用的数据集非常庞大，交互数据集有23G，大概6亿多条交互数据，用户有400多万，物品有一千多万。

刚开始处理这批数据的时候我的内心是崩溃的，一起不管在处理线上数据，还是我们私下训练各种推荐排序模型的时候，基本数据都没有到达过上亿的级别。这就意味着，可能这次比赛，召回层的作用会尤为重要。(推荐一般分为三个阶段，召回、排序和重排序，对于海量数据来说，召回意味着筛选，将物料池从百万级别降到百级别，这个工作听起来就很吓人了)同时，组委会还给了提示：

GNN这个东西以我们的软硬件实力，十有八九是用不起来了，所以，我们尝试了一下简单的思路，也就是本文的主题，ItemCF算法，而且我们没有优化它，甚至连参数都没有调整，只做了一些效率优化的工作。

我们采用pyspark完成整个数据的处理工作，首先，加载环境：

from pyspark import SparkContext

from pyspark import SparkConf

from pyspark.ml.feature import MinHashLSH

from pyspark.ml.linalg import Vectors

import pyspark.sql.functions as psf

from pyspark.sql import SparkSession

from pyspark.sql.types import FloatType

from pyspark.sql.types import StructType

from pyspark.sql.types import StringType

from pyspark.sql.types import StructField

from pyspark.sql.types import IntegerType

import math

import argparse

import time

import os

Path = "hdfs:///ECommAI/round2/train/"

spark = SparkSession \

.builder \

.appName("ItemCF") \

.master("yarn") \

.config("spark.executor.memory", "20g")\

.config("spark.executor.cores", "4")\

.config("spark.cores.max", "10")\

.config("spark.driver.memory","20g")\

.config("spark.executor.instances","10") \

.config("spark.sql.broadcastTimeout", "36000") \

.config("spark.local.dir", "/hadoop/spark") \

.config("spark.executor.extraJavaOptions", "-Xss80M") \

.getOrCreate()

sc = spark.sparkContext

包括了spark的上下文环境、MinHashLSH(用来加速计算近邻)、和一些基础库。数据在HDFS上。然后，加载数据(我们提前分好了训练集和测试集，但是为了省事就一起训练，得到线上提交结果)。

def PrepareTrainData():

train = spark.read.format("csv").option("delimiter", "\t").option("header", "false").schema(

actionSchema).load("/ECommAI/round2/train/*.csv")

ratingsRDD = train.rdd

return ratingsRDD

def PrepareTestData():

test = spark.read.format("csv").option("delimiter", "\t").option("header", "false").schema(

actionSchema).load("/ECommAI/round2/test/*.csv")

ratingsRDD = test.rdd

return ratingsRDD

train = PrepareTrainData()

test = PrepareTestData()

整合数据，计算物品相似度。训练数据以

的形式拿到，然后换位，聚合，得到

的数据，然后将每个

内的

转换成稀疏向量喂入MinHashLSH内，得到物品相似度矩阵，输出结果为

。

def get_feature_list_from_partition(iterator):

result = []

for arr in iterator:

arr_tmp = list(set(arr[1][1]))

arr_tmp.sort()

result.append((arr[0], Vectors.sparse(arr[1][0], arr_tmp, [1] * len(arr_tmp))))

return result

unionRDD = train.union(test)

userItemRDD = unionRDD.map(lambda x: (x[0], x[1]))

# 计算物品相似度

# 基于LSH的操作

max_user_id = userItemRDD.map(lambda x: int(x[0])).distinct().max()

itemUserRDD = userItemRDD.map(lambda x: (x[1], [x[0]])).reduceByKey(lambda x, y: x + y)

itemUserRDD = itemUserRDD.map(lambda x: (x[0], [max_user_id + 1, x[1]]))

item_vec_rdd = itemUserRDD.mapPartitions(get_feature_list_from_partition)

item_vec_df = item_vec_rdd.toDF(["item", "features"])

item_vec_df.show()

mh = MinHashLSH(inputCol="features", outputCol="hashes", numHashTables=5)

model = mh.fit(item_vec_df)

item_sim_df = model.approxSimilarityJoin(item_vec_df, item_vec_df, 0.98, distCol="JaccardDistance") \

.select(psf.col("datasetA.item").alias("idA"),

psf.col("datasetB.item").alias("idB"),

psf.col("JaccardDistance")).orderBy(psf.col("datasetA.item"), psf.col("JaccardDistance"))

item_sim_rdd = item_sim_df.rdd.filter(lambda x: x.idA != x.idB) \

.map(lambda x: (x.idA, x.idB, 1 - x.JaccardDistance))

rdd13 = item_sim_rdd.map(lambda data: ','.join(str(d) for d in data))

进入ItemCF的候选物品计算部分，构造形如

的相似度列表，每个物品获取20个近邻物品。

k = 20

unionRDD = unionRDD.distinct()

topkSim = item_sim_rdd.map(lambda x: (x[0], (x[0], x[1], x[2]))).groupByKey().map(

lambda x: (x[0], list(x[1]))).map(lambda x: (x[0], getTopN(x[1], k))).flatMap(

lambda x: x[1]).map(lambda x: (x[0], (x[1], x[2])))

根据ItemCF公式计算物品得分(此处公式就不写了，在黄老师的新书里，Spark构建协同过滤描述的很清楚)。

R2 = topkSim.join(unionRDD.map(lambda x: (x[1], (x[0], x[2]))))

R3 = R2.map(lambda x: ((x[1][1][0], x[1][0][0]), float(x[1][1][1]) * float(x[1][0][1])))

R4 = R3.reduceByKey(lambda x, y: x + y)

为每个用户，得到50个推荐结果，并存储结果。

N = 50

R5 = R4.leftOuterJoin(unionRDD.map(lambda x: ((x[0], x[1]), 1))).filter(lambda x: x[1][1] == None).map(

lambda x: (x[0][0], (x[0][0], x[0][1], x[1][0])))

R6 = R5.groupByKey().map(lambda x: (x[0], list(x[1])))

R7 = R6.map(lambda x: (x[0], getTopN(x[1], N)))

R9 = R7.map(toCSVLine)

R9.coalesce(1).saveAsTextFile("hdfs:///ECommAI/round2/ItemCF/predict/" + str(int(time.time())))

中间的一些功能性函数：

def PrePareBestK(sc, Path):

K = sc.textFile(Path)

K = int(K.collect()[0])

return K

def find(user, user_list):

for u in user_list:

if u == user:

return True

return False

def takeSecond(elem):

return elem[2]

def getTopN(x, k):

x.sort(key=takeSecond, reverse=True)

x = x[:k]

return x

def toCSVLine(data):

output_str = str(data[0]) + "\t"

return output_str + ','.join([str(d[1]) for d in data[1]])

总的来说，就是一次基于pyspark的ItemCF线下实战，数据集用的天池比赛的，整体运行时间在8个小时左右，结果是0.01569(还在第一页)。

基于传统的推荐算法，可能上限就到这了，想得到更好的结果，可能采用一些基于深度学习的方法可能更好，我们也尝试了基于Word2Vec词向量的itemCF，YoutubeMatchModel、DeepFM方法，甚至还考虑过用关联规则而不是GNN来补充交互数据，不过都是后话了。