pyspark-topnproducts.py

def topNProducts(rec, topN):
  x = [ ]
  x = list(sorted(rec, key=lambda k: float(k.split(",")[4]), reverse=True))
  import itertools
  return (y for y in list(itertools.islice(x, 0, topN)))

products = sc.textFile("/public/retail_db/products")
productsFiltered = products.filter(lambda rec: rec.split(",")[4] != "")

for i in productsFiltered.\
map(lambda rec: (int(rec.split(",")[1]), rec)).\
groupByKey().\
flatMap(lambda rec: topNProducts(rec[1], 5)).\
collect():
  print(i)