在一次采访中，我被问及一个bigdata问题，其中给出了具有以下模式的数据集

UserId, MovieId, Rating

这里每一行都有用户对电影数据的评分。因此，根据用户观看的电影给出分级（分级基于用户观看的时间，或者其他一些无关紧要的标准）。
问题是要为每个评分超过7的用户获取最高评分的movieid列表，而这个用户还没有观看。或者基本上是可以推荐给netflix用户的电影列表。
例如下面的数据集

User_123    Movie_442   5
User_123    Movie_434   8
User_123    Movie_487   6
User_123    Movie_423   9
User_415    Movie_442   8
User_415    Movie_994   9
User_993    Movie_884   7
User_993    Movie_887   6
User_993    Movie_883   9

我正在寻找以下输出：-

User_123    Movie_883
User_123    Movie_994
User_415    Movie_423
User_415    Movie_434
User_415    Movie_883
User_993    Movie_423
User_993    Movie_434
User_993    Movie_442
User_993    Movie_994

我有解决这个问题的办法 Apache Pig 但我正在寻找更优化的方法。有人能给我一个更好的解决办法吗
loadfile=使用pigstorage（'\t'）作为（userid:chararray，movieid:chararray，额定值：int）；useridfield=foreach loadfile生成userid；distinctusers=不同的useridfield；ratinggreaterthan7=按评级>7筛选加载文件；usersmovielist=foreach ratinggreater than7生成userid，movieid；crossjoin=cross distinctusers，usersmovielist；filtermovies=filter crossjoin by not（distinctusers:：userid==usersmovielist:：userid）；usersmovie=foreach filtermovies生成distinctusers:：userid作为userid，usersmovielist:：movieid作为movieid；filteralreadywatched=join usersmovie by（userid，movieid）left outer，loadfile by（userid，movieid）；filteralreadywatched1=筛选器filteralreadywatched by（loadfile:：userid为空，loadfile:：movieid为空）；usermoviewatchlist=foreach filteralreadywatched1生成usersmovie:：userid，usersmovie:：movieid；使用pigstorage（'\t'）将usermoviewatchlist存储到'$output'中；