ratings= sqlContext.read.csv('/FileStore/tables/qthksaj91493514788924/rating_cleaned.csv', sep=',' , inferSchema=True, header=True)
anime = sqlContext.read.csv('/FileStore/tables/ezjvoa7k1493514682679/anime_clean.tsv', sep='\t', inferSchema=True, header=True)
names= sqlContext.read.csv('/FileStore/tables/p4kjul861493612915497/anime_name.csv', inferSchema=True, header=True)

animes = spark.createDataFrame(anime.rdd.map(lambda x: (x[0], x[1].lower().replace('"',"").replace(' ',"").split(','))), ['anime_id','genre'])
print animes.show(2)
print anime.count()

count  = []
for i in animes.collect():
  count.extend(i[1])
print len(count), len(set(count))
count_genre = len(set(count))

from pyspark.ml.feature import CountVectorizer
from pyspark.ml.linalg import Vectors

cv = CountVectorizer(inputCol="genre", outputCol="features", vocabSize=count_genre, minDF=2.0)
cvmodel = cv.fit(animes)
animes_transformed = cvmodel.transform(animes)
animes_transformed.take(5)

fnldata = spark.createDataFrame(animes_transformed.select('anime_id', 'features').rdd.map(lambda x: (x[0], Vectors.dense(x[1]))), ['id', 'DenseVector'])
print fnldata.take(5)
print fnldata.show(5)
fnldata.cache()

# Get distance functions from Sklearn
from sklearn.metrics.pairwise import cosine_similarity
from sklearn.metrics import jaccard_similarity_score
from sklearn.metrics.pairwise import euclidean_distances
from sklearn.metrics.pairwise import manhattan_distances
from pprint import pprint
import pandas as pd

test_id = 22043
test_vector= fnldata.rdd.lookup(test_id)

cosine_dist =spark.createDataFrame(fnldata.rdd.map(lambda x: (x[0], float(cosine_similarity(x[1], test_vector)[0][0]))), ['anime_id', 'cosine_sim'])

cosine_recomm=cosine_dist.join(names, names.anime_id==cosine_dist.anime_id).sort('cosine_sim',ascending=False).take(26)
print pd.DataFrame(cosine_recomm).head(26)

#Calculate Cosine Similarity between the test id and rdd elements
euclidean_dist =spark.createDataFrame(fnldata.rdd.map(lambda x: (x[0], float( euclidean_distances(x[1], test_vector)[0][0]))), ['anime_id', 'euclidean_distances'])

# take 26 of the values for recommendation
euclidean_recomm=euclidean_dist.join(names, names.anime_id==euclidean_dist.anime_id).sort('euclidean_distances',ascending=True).take(26)
#pprint(euclidean_recomm)
pd.DataFrame(euclidean_recomm).head(25)

#Calculate Manhattan Dist between the test id and rdd elements
manhattan_dist =spark.createDataFrame(fnldata.rdd.map(lambda x: (x[0], float( manhattan_distances(x[1], test_vector)[0][0]))), ['anime_id', 'manhattan_distances'])

# take 26 of the values for recommendation
manhattan_recomm=manhattan_dist.join(names, names.anime_id==manhattan_dist.anime_id).sort('manhattan_distances',ascending=True).take(26)
#pprint(manhattan_recomm)
pd.DataFrame(manhattan_recomm).head(26)

jaccard_sim =spark.createDataFrame(fnldata.rdd.map(lambda x: (x[0], float(jaccard_similarity_score( test_vector[0], x[1])))), ['anime_id', 'jaccard_similarity'])

# take 26 of the values for recommendation
jaccard_recomm=jaccard_sim.join(names, names.anime_id==jaccard_sim.anime_id).sort('jaccard_similarity',ascending=False).take(26)
#pprint(jaccard_recomm)
pd.DataFrame(jaccard_recomm).head(25)

#print the recommedations
print "Cosine , Euclidean, Manhattan, Jaccard"
fnl_recomm=[]
for i in range(len(cosine_recomm)) :
  print (cosine_recomm[i][3],  euclidean_recomm[i][3], manhattan_recomm[i][3], jaccard_recomm[i][3])
  fnl_recomm.append(cosine_recomm[i][3])
  fnl_recomm.append(euclidean_recomm[i][3])
  fnl_recomm.append(manhattan_recomm[i][3])
  fnl_recomm.append(jaccard_recomm[i][3])


print len(fnl_recomm), len(set(fnl_recomm))

pprint (set(fnl_recomm))

fnldata.rdd.cartesian(fnldata.rdd).map(lambda (x, y):(x[0], y[0], cosine_similarity(x[1], y[1]))).take(5)

fnldata.rdd.cartesian(fnldata.rdd).map(lambda (x, y):(x[0], y[0], float(cosine_similarity(x[1], y[1])[0][0]), jaccard_similarity_score(x[1], y[1]), manhattan_distances(x[0], y[0])[0][0], euclidean_distances(x[0], y[0])[0][0])).take(5)

recommnd = spark.createDataFrame(fnldata.rdd.cartesian(fnldata.rdd).map(lambda (x, y):(x[0], y[0], float(cosine_similarity(x[1], y[1])[0][0]), float(jaccard_similarity_score(x[1], y[1])), float(manhattan_distances(x[0], y[0])[0][0]), float(euclidean_distances(x[0], y[0])[0][0]))), ['anime_id1', 'anime_id2','cosine_similarity', 'jaccard_similarity', 'manhattan_distances', 'euclidean_distances'])
print recommnd.show(25)

recommnd.filter(recommnd.anime_id1==28977).show()



StarcoderdataJupyterScriptsDedupFiltered