Big Data

HDFS

POSIX Commands

HDFS Specific Commands

HDFS Configuration Files

Recommendation Engine

Implicit rating

HDFS

POSIX Commands

hdfs dfs -#arg
- ls #List home directory
- ls . #List current directory
- ls / #List root directory
- cp #Copy files
  - file:///dir
  - hdfs://server:port_num/dir
- cat #Display content of file
- chgrp
- chmod
- chown
- du
- mkdir #Create directory under home directory
- mv
- rm #Remove directory and contents
- stat
- tail

HDFS Specific Commands

copyFromLocal OR put
copyToLocal OR get
getmerge #Get files in directors and merge them into single local file
setrep #Override default level of replication (-R for entire tree)
hdfs dfsadmin -report #Check health status of cluster (e.g. amount of space on each DataNode)

HDFS Configuration Files

https://youtu.be/LtGliUam-_U

core-site.xml #Parameters for HDFS and MapReduce
hdfs-site.xml #Parameters for HDFS daemons (NameNode and DataNodes)
mapred-site.xml #Parameters for MapReduce daemons, JobTracker and TaskTrackers

JAVA_HOME #Set to java SDK

HADOOP PROGRAMS

Pig

pig #script dir #Run script

Hive

hive -f #script dir #Run script
hive -e 'SELECT * FROM #table' #Run inline command

Flume

Sqoop

Ozzie

PySpark

Recommendation Engine

ALS
- Feature discovery
- Item grouping
- Dimensionality reduction
- Image compression

#df.show() #Show dataframe
#rating_df.groupBy("#Genre").sum().show()
#ratings.filter(col("#num_plays") > 0).groupBy("#userId").count().select(avg("count")).show()

#ratings.printSchema() #See the datatypes of the ratings dataset
#ratings = #ratings.select(#ratings.#userId.cast("#integer"), #ratings.#movieId.cast("#integer"), #ratings.#rating.cast("#double")) #Tell Spark to convert the columns to the proper data types

from pyspark.sql.functions import col
#ratings.filter(col("#userId") < #100).show() #Filter to show only userIds less than 100
OR #ratings.groupBy("#userId").count().show() #Group data by userId, count ratings
#ratings.groupBy("#movieId").count().select(#avg("#count")).show() #Calculate average rating per movie

from pyspark.sql.functions import monotonically_increasing_id #Import monotonically_increasing_id
#ratings = to_long(#R) #Use the to_long() function to convert the dataframe to the "long" format.

#users = #ratings.select("#user col name").distinct().coalesce(1) #Get unique users and repartition to 1 partition
#users = #users.withColumn("#userId", monotonically_increasing_id()).persist() #Create a new column of unique integers called "userId" in the users dataframe and ensure userId remains the same throughout
#movies= #ratings.select("#movie col name").distinct().coalesce(1) #Get unique movies and repartition to 1 partition
#movies= #movies.withColumn("#movieId", monotonically_increasing_id()).persist() #Create a new column of unique integers called "movieId" in the movies dataframe and ensure movieId remains the same throughout
#crossjoin = #users.crossJoin(#movies) #Can perform cross-join
#movie_ratings = #ratings.join(#users, "#User", "left").join(#movies, "#Movie", "left")#.fillna(0) #Join the ratings, users and movies dataframes

sparsity = (1.0 - (#ratings.select("#rating").count()*1.0) / (#ratings.select("#userId").distinct().count() * #ratings.select("#movieId").distinct().count()))*100 #Calculate sparsity
print("The ratings dataframe is ", "%.2f" % sparsity + "% empty.")

(#training_data, #test_data) = #ratings.randomSplit([#0.8, #0.2], seed=1234) #Split the ratings dataframe into training and test data

from pyspark.ml.recommendation import ALS
#als = ALS(userCol = "#userId", itemCol = "#movieId", ratingCol = "#rating", rank = #10, maxIter = #15, regParam = #0.1, coldStartStrategy = "#drop", nonnegative = #True, implicitPrefs = #False) #Set the ALS hyperparameters (rank: latent features; regParam: lambda to prevent overfitting; alpha: used with implicit rating for confidence; nonnegative: ensure positive values; coldStartStrategy: use users with ratings found in both training and test set for RMSE calculation; implicitPrefs: indicate type of rating)

from pyspark.ml.tuning import ParamGridBuilder, CrossValidator
param_grid = ParamGridBuilder() \
.addGrid(als.rank, [10, 50, 100, 150]) \
.addGrid(als.maxIter, [5, 50, 100, 200]) \
.addGrid(als.regParam, [.01, .05, .1, .15]) \
.build() #Add hyperparameters and their respective values to param_grid

#model = #als.fit(#training_data) #Fit the model to the training_data
#test_predictions = #model.transform(#test_data) #Generate predictions on the test_data
#test_predictions.show()

from pyspark.ml.evaluation import RegressionEvaluator #Import RegressionEvaluator to calculate RMSE
#evaluator = RegressionEvaluator(metricName = "#rmse", labelCol = "#rating", predictionCol = "#prediction")
#cv = CrossValidator(estimator = #als, estimatorParamMaps = #param_grid, evaluator = #evaluator, numFolds = #5) #Build cross validation using CrossValidator
#model = #cv.fit(#train) #Fit cross validator to the 'train' dataset
#best_model = #model.bestModel #Extract best model from the cv model above
best_model.getRank()
best_model.getMaxIter()
best_model.getRegParam()

#RMSE = #evaluator.evaluate(#test_predictions) #Evaluate the "test_predictions" dataframe for explicit rating, return value is x rating score above or below values original ratings matrix
getRMSEs(#[matrix], #actual) #Compute RMSE

#original_ratings.filter(col("#userId") == #63).sort("#rating", ascending = False).show() #Look at the movie for user 63
#recommendations.filter(col("#userId") == #63).show() #Look at the movies recommended to user 63

Implicit rating

ranks = [10, 20, 30, 40]
maxIters = [10, 20, 30, 40]
regParams = [.05, .1, .15]
alphas = [20, 40, 60, 80]

for r in ranks:
for mi in maxIters:
for rp in regParams:
for a in alphas:
model_list.append(ALS(userCol= "userId", itemCol= "songId", ratingCol= "num_plays", rank = r, maxIter = mi, regParam = rp, alpha = a, coldStartStrategy="drop", nonnegative = True, implicitPrefs = True)) #For loop will automatically create and store ALS models
len(model_list) == (len(ranks)*len(maxIters)*len(regParams)*len(alphas)) #Validate

(#training, #test) = #msd.randomSplit([#0.8, #0.2]) #Split the data into training and test sets
train1, train2, train3, train4, train5 = training.randomSplit([0.2, 0.2, 0.2, 0.2, 0.2], seed = 1) #Building 5 folds within the training set
fold1 = train2.union(train3).union(train4).union(train5)
fold2 = train3.union(train4).union(train5).union(train1)
fold3 = train4.union(train5).union(train1).union(train2)
fold4 = train5.union(train1).union(train2).union(train3)
fold5 = train1.union(train2).union(train3).union(train4)

foldlist = [(fold1, train1), (fold2, train2), (fold3, train3), (fold4, train4), (fold5, train5)]

ROEMS = [] #Create empty list to fill with ROEMs from each model

for model in model_list: #Loops through all models and all folds
for ft_pair in foldlist:
# Fits model to fold within training data
fitted_model = model.fit(ft_pair[0])
# Generates predictions using fitted_model on respective CV test data
predictions = fitted_model.transform(ft_pair[1])
# Generates and prints a ROEM metric CV test data
r = ROEM(predictions)
print ("ROEM: ", r)

v_fitted_model = model.fit(training) #Fits model to all of training data and generates preds for test data
v_predictions = v_fitted_model.transform(test)
v_ROEM = ROEM(v_predictions) #Generate Rank Ordering Error Metric (ROEM) to check if the rank order of the prediction is similar to actual rank

ROEMS.append(v_ROEM) #Adds validation ROEM to ROEM list
print ("Validation ROEM: ", v_ROEM)

i = numpy.argmin(ROEMS)
print("Smallest ROEM: ", ROEMS[i])
best_model.getRank()
best_model.getMaxIter()
best_model.getRegParam()
best_model.getAlpha()

from pyspark.sql.functions import col # Import the col function

#binary_test_predictions.show() # Look at the test predictions

ROEM(#binary_test_predictions) # Evaluate ROEM on test predictions

#binary_test_predictions.filter(col("#userId") == #42).show() # Look at user 42's test predictions

Item Weighting: More views = Higher weight
User Weighting: More movies = Lower weight to unseen movies as users know what they want

Google Sites

Report abuse