<jupyter_start><jupyter_text>PySpark Recommendation using Apache SystemML<jupyter_code>%load_ext autoreload
%autoreload 2
%matplotlib inline

import numpy as np
import matplotlib.pyplot as plt
plt.rcParams['figure.figsize'] = (10, 6)<jupyter_output><empty_output><jupyter_text>Data Download the dataset<jupyter_code>%%sh
curl -O http://snap.stanford.edu/data/amazon0601.txt.gz
gunzip amazon0601.txt.gz<jupyter_output>% Total    % Received % Xferd  Average Speed   Time    Time     Time  Current
                                 Dload  Upload   Total   Spent    Left  Speed

  0     0    0     0    0     0      0      0 --:--:-- --:--:-- --:--:--     0
  0 11.2M    0 14223    0     0  50034      0  0:03:55 --:--:--  0:03:55 51532
 39 11.2M   39 4510k    0     0  3570k      0  0:00:03  0:00:01  0:00:02 3594k
100 11.2M  100 11.2M    0     0  7196k      0  0:00:01  0:00:01 --:--:-- 7233k<jupyter_text>Load the data and prepare it for the processing.<jupyter_code>import pyspark.sql.functions as F
dataPath = "amazon0601.txt"

X_train = (sc.textFile(dataPath)
    .filter(lambda l: not l.startswith("#"))
    .map(lambda l: l.split("\t"))
    .map(lambda prods: (int(prods[0]), int(prods[1]), 1.0))
    .toDF(("prod_i", "prod_j", "x_ij"))
    .filter("prod_i < 500 AND prod_j < 500") # Filter for memory constraints
    .cache())

max_prod_i = X_train.select(F.max("prod_i")).first()[0]
max_prod_j = X_train.select(F.max("prod_j")).first()[0]
numProducts = max(max_prod_i, max_prod_j) + 1 # 0-based indexing
print("Total number of products: {}".format(numProducts))<jupyter_output>Total number of products: 500<jupyter_text>SystemML - Poisson Nonnegative Matrix Factorization (PNMF) Create SystemML MLContext<jupyter_code>from SystemML import MLContext
ml = MLContext(sc)<jupyter_output><empty_output><jupyter_text>Define PNMF kernel in SystemML's DML using the R-like syntax for PNMF<jupyter_code>pnmf = """
# data & args
X = read($X)
X = X+1 # change product IDs to be 1-based, rather than 0-based
V = table(X[,1], X[,2])
size = ifdef($size, -1)
if(size > -1) {
    V = V[1:size,1:size]
}
max_iteration = as.integer($maxiter)
rank = as.integer($rank)

n = nrow(V)
m = ncol(V)
range = 0.01
W = Rand(rows=n, cols=rank, min=0, max=range, pdf="uniform")
H = Rand(rows=rank, cols=m, min=0, max=range, pdf="uniform")
losses = matrix(0, rows=max_iteration, cols=1)

# run PNMF
i=1
while(i <= max_iteration) {
  # update params
  H = (H * (t(W) %*% (V/(W%*%H))))/t(colSums(W)) 
  W = (W * ((V/(W%*%H)) %*% t(H)))/t(rowSums(H))
  
  # compute loss
  losses[i,] = -1 * (sum(V*log(W%*%H)) - as.scalar(colSums(W)%*%rowSums(H)))
  i = i + 1;
}

# write outputs
write(losses, $lossout)
write(W, $Wout)
write(H, $Hout)
"""<jupyter_output><empty_output><jupyter_text>Run the PNMF script on SystemML with Spark<jupyter_code>ml.reset()
outputs = ml.executeScript(pnmf, {"X": X_train, "maxiter": 100, "rank": 10}, ["W", "H", "losses"])<jupyter_output><empty_output><jupyter_text>Plot training loss over time<jupyter_code>losses = outputs.getDF(sqlContext, "losses")
xy = losses.sort(losses.ID).map(lambda r: (r[0], r[1])).collect()
x, y = zip(*xy)
plt.plot(x, y)
plt.xlabel('Iteration')
plt.ylabel('Loss')
plt.title('PNMF Training Loss')<jupyter_output><empty_output>
{'fraction_non_alphanumeric': 0.17040910489080283, 'fraction_numerical': 0.05259920024607813, 'mean_word_length': 4.003076923076923, 'pattern_counts': {'":': 3, '<': 30, '<?xml version=': 0, '>': 28, 'https://': 0, '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': '7', 'dir': 'jupyter-structured-clean-dedup', 'id': '654437', 'license': 'MIT', 'n_tokens_mistral': 1434, 'n_tokens_neox': 1319, 'n_words': 293, 'path': 'PySpark_Recommendation.ipynb', 'repo_name': 'irfanalidv/ML_with_Apache_SystemML', 'size': '3251'}