DEV Community: Akın

Building an Anime Recommendation System with PySpark in SageMaker

Akın — Sun, 17 Mar 2024 10:35:27 +0000

Demonstration of an Anime Recommendation System with PySpark in SageMaker `v1.0.0-alpha01`

Understanding the Dataset

Our journey begins with understanding the dataset. We will be using the MyAnimeList dataset sourced from Kaggle, which contains valuable information about anime titles, user ratings, and more. This dataset will serve as the foundation for our recommendation system.

Preprocessing the Data

Before diving into model building, we will preprocess the dataset to ensure it is clean and structured for analysis. While the preprocessing steps have already been completed, we will briefly discuss the importance of data preprocessing in the context of recommendation systems.

for detailed preprocessing check out this notebook
anime-recommendation-system.ipynb

TODO: Add all preprocessing step-by-step with explanations

Visualization

Visualizing the preprocessed data can provide valuable insights into the distribution of anime ratings, user preferences, and other patterns. We will utilize various visualization techniques to gain a better understanding of our dataset.

You can find data visualization techniques applied to better understand the content of the data in anime-recommendation-system.ipynb. You can find one of them below as example.

TODO: Add all visualizations step-by-step with explanations and insights

Recommendation System

We employed the Alternating Least Squares (ALS) algorithm to build the recommendation system using PySpark. We used the mean square error algorithm to find the best model. We calculated how accurately it measured the real value by training the data allocated for the test and estimating the remaining data.

Since these parameters had the lowest mean square error, we created a model with these parameters to obtain more accurate results.

rank, iter, lambda_ = 50, 10, 0.1
model = ALS.train(rating, rank=rank, iterations=iter, lambda_=lambda_, seed=5047)

Cloud Infrastructure Setup with Terraform

First of all we need to upload preproocessed data to a s3 bucket. Use the following commands. You can reach to upload.py in there.

$ cd sagemaker
$ python upload_data.py -n 'anime-recommendation-system' -r 'eu-central-1' -f '../preprocessed_data'

../preprocessed_data\user.csv  4579798 / 4579798.0  (100.00%)00%)

After that run terraform commands to create an Amazon SageMaker Notebook Instance:

instance.tf

$ terraform init
Initializing the backend...

Initializing provider plugins...
- Finding latest version of hashicorp/aws...
- Installing hashicorp/aws v5.41.0...
- Installed hashicorp/aws v5.41.0 (signed by HashiCorp)
...
$ terraform plan
...
Plan: 4 to add, 0 to change, 0 to destroy.
$ terraform apply --auto-approve
aws_iam_policy.sagemaker_s3_full_access: Creating...
aws_iam_role.sagemaker_role: Creating...
aws_iam_policy.sagemaker_s3_full_access: Creation complete after 1s [id=arn:aws:iam::749270828329:policy/SageMaker_S3FullAccessPoliciy]
aws_iam_role.sagemaker_role: Creation complete after 1s [id=AnimeRecommendation_SageMakerRole]
aws_iam_role_policy_attachment.sagemaker_s3_policy_attachment: Creating...
aws_sagemaker_notebook_instance.notebookinstance: Creating...
aws_iam_role_policy_attachment.sagemaker_s3_policy_attachment: Creation complete after 1s [id=AnimeRecommendation_SageMakerRole-20240317120654686300000001]
aws_sagemaker_notebook_instance.notebookinstance: Still creating... [10s elapsed]
...
Apply complete! Resources: 4 added, 0 changed, 0 destroyed.

Create a new conda_python3 notebook and Run `sagemaker-anime-recommendation-system.ipynb` step by step on the Notebook Instance

You can look at the sagemaker-anime-recommendation-system-test.html file to see the output.

In this code, model data is saved locally and then uploaded properly to an s3 bucket.

model.save(SparkContext.getOrCreate(), 'model')

import boto3
import os

# Initialize S3 client
s3 = boto3.client('s3')

# Upload files to the created bucket
bucketname = 'anime-recommendation-system'
local_directory = './model'
destination = 'model/'
for root, dirs, files in os.walk(local_directory):
    for filename in files:
        # construct the full local path
        local_path = os.path.join(root, filename)

        relative_path = os.path.relpath(local_path, local_directory)
        s3_path = os.path.join(destination, relative_path)

        s3.upload_file(local_path, bucketname, s3_path)

To load and test the model, run the `test_another_instance.ipynb` jupyter notebook

In this code, model data is retrieved from the s3 bucket and loaded using the Matrix Factorization Model

import boto3
import os 

s3_resource = boto3.resource('s3')
bucket = s3_resource.Bucket('anime-recommendation-system') 
for obj in bucket.objects.filter(Prefix = 'model'):
    if not os.path.exists(os.path.dirname(obj.key)):
        os.makedirs(os.path.dirname(obj.key))
    bucket.download_file(obj.key, obj.key)

from pyspark import SparkContext
from pyspark.mllib.recommendation import MatrixFactorizationModel

m = MatrixFactorizationModel.load(SparkContext.getOrCreate(), 'model')

As a result, I explained how to create the model by loading the model data into each separate environment, making the model easier to use, and processing it with SageMaker's notebook instance. See you in another article...👋👋

My Huawei Cloud Practicum Experience

Akın — Sat, 07 Jan 2023 18:13:18 +0000

This article is about my experience in the Huawei Cloud Practicum organized by Patika, sponsored by Huawei Cloud.

Intro
First Case
Homeworks
Final Project
Conclusion

Intro

I am Akın. I am developing myself in cloud computing. During this process, I wanted to add a hands-on experience to my learning process by participating in this practicum. In this blog post, I will describe to you what I did during this process.

This practicum was a 6-week process. To participate in this process, the given case was asked to be done as desired.

First Case

We were asked to show how communication is provided by establishing peering connections between the VPCs. Since the case is simple, I created a challenge for myself. Around this time, I was starting to learn Terraform. That's why I wanted to add a hands-on experience to the learning process and develop the infrastructure as code with Terraform.

My first job was to review the code samples on the web and in the terraform-provider-huaweicloud repository and understand the variables in Terraform. After a week, I completed the case. I brought together what I did during the process. You can reach my blog post and terraform codes from these links.

Homeworks

During the first two weeks, we completed five exercises in Koolabs. With these exercises, we started to experience and learn Huawei Cloud services with a hands-on approach. In the third week, we did an exercise aiming to use the CCE (Kubernetes) cluster basic features. With this exercise, we experienced the CCE service hands-on.

Final Project

In the final project, I converted my monolithic survey management system application, which I had previously developed, into serverless architecture and tried to adapt it to the cloud. You can find my project codes here.

Creation of serverless functions
Creating a pipeline with DevCloud's CloudPipeline
Serverless backend with FunctionGraph
Triggering functions with API Gateway
Static website hosting with OBS

1. Creation of serverless functions

First, I created the internal application logic and serverless functions by separating the API handlers in the monolith app into events.

Structure of the project

survey-builder
 ┣ functions
 ┃ ┣ register
 ┃ ┃ ┗ main.go
 ┃ ┣ ...
 ┃ ┗ ...
 ┣ internal
 ┃ ┣ database
 ┃ ┣ go-runtime
 ┃ ┣ handler
 ┃ ┣ model
 ┃ ┗ service
 ┣ vendor
 ┣ go.mod
 ┗ go.sum

Below is a serverless function example of register event. I handled APIG Trigger events using Huawei Cloud's FunctionGraph SDK. I used GaussDB for NoSQL for the database.

package main

import (
    "bytes"
    "encoding/base64"
    "encoding/json"
    "net/http"

    "huaweicloud.com/akinbe/survey-builder-app/internal/go-runtime/events/apig"
    "huaweicloud.com/akinbe/survey-builder-app/internal/go-runtime/go-api/context"
    "huaweicloud.com/akinbe/survey-builder-app/internal/go-runtime/pkg/runtime"
    "huaweicloud.com/akinbe/survey-builder-app/internal/handler"
)

func RegisterHandler(payload []byte, ctx context.RuntimeContext) (interface{}, error) {
    // Handle Apig Trigger Event Request
    var apigEvent apig.APIGTriggerEvent
    err := json.Unmarshal(payload, &apigEvent)
    if err != nil {
        apigResp := apig.APIGTriggerResponse{
            Body: err.Error(),
            Headers: map[string]string{
                "content-type": "application/json",
            },
            StatusCode: http.StatusBadRequest,
        }
        return apigResp, nil
    }

    // Parse the 'data' value from the trigger event, which takes the 'data' value
    // from the request sent by the client and carries it to the backend.
    // Then, decode the base64 encoded data.
    data, _ := base64.StdEncoding.DecodeString(apigEvent.PathParameters["data"])
    data_str := bytes.NewBuffer(data).String()

    // Event Logic
    viewuser, status, err := handler.UserSignupPostHandler(data_str)
    if err != nil {
        apigResp := apig.APIGTriggerResponse{
            Body: err.Error(),
            Headers: map[string]string{
                "content-type": "application/json",
            },
            StatusCode: status,
        }
        return apigResp, nil
    } else {
        user, _ := json.Marshal(viewuser)
        user_str := string(user)
        apigResp := apig.APIGTriggerResponse{
            Body: user_str,
            Headers: map[string]string{
                "content-type": "application/json",
            },
            StatusCode: http.StatusOK,
        }
        return apigResp, nil
    }
}

func main() {
    runtime.Register(RegisterHandler)
}

2. Creating a pipeline with DevCloud's CloudPipeline

With DevCloud's Cloud Pipeline, I built the functions with docker and then transferred the images to SWR. With the container image option in FunctionGraph, I created the functions by pulling them from SWR. You can view the Dockerfile of the register event below.

FROM golang:1.19-alpine AS builder

RUN apk add --update --no-cache gcc git build-base

WORKDIR /src
COPY . /src
RUN CGO_ENABLED=0 go build -o /bin/register

FROM scratch
COPY --from=builder /bin/register /bin/register
ENTRYPOINT ["/bin/register"]

I tested the functions in FunctionGraph. However, I resorted to another method because I was constantly getting the below error and could not find a solution.

function invocation exception, error: CrashLoopBackOff: The application inside the container keeps crashing

I built and packaged the functions with the bash script below.

#! /bin/bash

for dir in functions/*/; do
  # Extract the function name from the directory path
  function_name=$(basename "$dir")
  cd ${GOPATH}/src/huaweicloud.com/akinbe/survey-builder/functions/$function_name
  package="${function_name}_go1.x.zip"
  go build -o handler main.go
  zip $package handler
done

With a Cloud Build task in DevCloud's Cloud Pipeline, I transferred the zip files to OBS (Object Storage Service). I created the functions in FunctionGraph by pulling the zip files from OBS. And then, I tested all functions successfully with APIG events.

3. Serverless backend with FunctionGraph

Below is the test case of the register function. It returned an HTTP status of 500 because there is currently no database connection.

4. Triggering serverless functions with API Gateway

In API Gateway, I created my APIs in an API Group where each endpoint corresponds to a function. I have configured the backend of the APIs with the functions in the Function Graph.

Below is the test of the register event at the /api/v1/signup endpoint.

5. OBS Static Web Site Hosting

I hosted my frontend files in OBS with static website hosting. However, I could not perform a full presentation since I received 405 not allowed responses from the POST requests sent from the client to the APIs.

Conclusion

As a result, I tried to implement a project with the Cloud first approach. I implemented a cloud-adapted project using the most appropriate services at every step. With its pros and cons, I tried to create an automated system in DevCloud's Cloud Pipeline, serverless architecture in FunctionGraph, and a gateway that serves backend logic to clients in API Gateway.

In this process, I had the chance to get to know and use Huawei Cloud services hands-on. My thanks to everyone who made this process possible.
See you in my next blog posts... 👋👋