code:

# Set up the Google Colab environment for pyspark, including integration with Google Drive

from google.colab import drive
drive.mount('/content/drive')

!pip install pyspark

from pyspark.sql import SparkSession
from pyspark.sql.functions import col, count, row_number, split, explode
from pyspark.sql.window import Window
import unittest

# Initialize the Spark Session for Google Colab
def initialize_spark_session():
    """Initialize and return a Spark session configured for Google Colab"""
    return SparkSession.builder\
        .master("local")\
        .appName("Colab")\
        .config('spark.ui.port', '4050')\
        .getOrCreate()

"""
This code will ingest data from the MovieLens 20M dataset. The goal will be to use Spark to find the following:

- The most common tag for each movie title:
  Utilizing Spark's DataFrame operations, the script joins the movies dataset with the tags dataset on the movie ID.
  After joining, it aggregates tag data by movie title to count occurrences of each tag, then employs a window function
  to rank tags for each movie. The most frequent tag for each movie is identified and selected for display.

- The most common genre rated by each user:
  Similarly, the script joins the movies dataset, which includes genre information, with the ratings dataset on the movie ID.
  It then groups the data by user ID and genre to count how many times each user has rated movies of each genre.
  A window function is again used to rank the genres for each user based on these counts. The top genre for each user
  is then extracted and presented.


"""

def load_dataset(spark, file_path, has_header=True, infer_schema=True):
    """Load the need files, with some error checking for good measure"""
    try:
        df = spark.read.csv(file_path, header=has_header, inferSchema=infer_schema)
        if df.head(1):
            return df
        else:
            raise ValueError("DataFrame is empty, check the dataset or path.")
    except Exception as e:
        raise IOError(f"Failed to load data: {e}")

def calculate_most_common_tag(movies_df, tags_df):
    """Calculates the most common tag for each movie title."""
    try:
        movies_tags_df = movies_df.join(tags_df, "movieId")
        tag_counts = movies_tags_df.groupBy("title", "tag").agg(count("tag").alias("tag_count"))
        window_spec = Window.partitionBy("title").orderBy(col("tag_count").desc())
        most_common_tags = tag_counts.withColumn("rank", row_number().over(window_spec)) \
                             .filter(col("rank") == 1) \
                             .drop("rank") \
                             .orderBy(col("tag_count").desc())
        return most_common_tags
    except Exception as e:
        raise RuntimeError(f"Error calculating the most common tag: {e}")

def calculate_most_common_genre(ratings_df, movies_df):
    """Calculates the most common genre rated by each user."""
    try:
        movies_df = movies_df.withColumn("genre", explode(split(col("genres"), "[|]")))
        ratings_genres_df = ratings_df.join(movies_df, "movieId")
        genre_counts = ratings_genres_df.groupBy("userId", "genre").agg(count("genre").alias("genre_count"))
        window_spec = Window.partitionBy("userId").orderBy(col("genre_count").desc())
        most_common_genres = genre_counts.withColumn("rank", row_number().over(window_spec)) \
                                 .filter(col("rank") == 1) \
                                 .drop("rank") \
                                 .orderBy(col("genre_count").desc())
        return most_common_genres
    except Exception as e:
        raise RuntimeError(f"Error calculating the most common genre: {e}")

def main():
    # Calling Spark session info
    spark = initialize_spark_session()
    # Load CSV files from MovieLens
    movies_df = load_dataset(spark, "/content/drive/My Drive/spark_project/movies.csv")
    tags_df = load_dataset(spark, "/content/drive/My Drive/spark_project/tags.csv")
    ratings_df = load_dataset(spark, "/content/drive/My Drive/spark_project/ratings.csv")

    # Perform analysis
    most_common_tag = calculate_most_common_tag(movies_df, tags_df)
    most_common_genre = calculate_most_common_genre(ratings_df, movies_df)

    # Displaying results
    print("Most Common Tag for Each Movie Title:")
    most_common_tag.show()
    print("Most Common Genre Rated by User:")
    most_common_genre.show()

class TestMovieLensAnalysis(unittest.TestCase):
    '''This will perform some unittests on a truncated set of files in a dedicated testing directory'''
    def setUp(self):
        self.spark = initialize_spark_session()
        self.movies_path = "/content/drive/My Drive/spark_testing/movies.csv"
        self.tags_path = "/content/drive/My Drive/spark_testing/tags.csv"
        self.ratings_path = "/content/drive/My Drive/spark_testing/ratings.csv"

    def test_load_dataset(self):
        # Assuming fake paths to simulate failure
        with self.assertRaises(IOError):
            load_dataset(self.spark, "fakepath.csv")

    def test_calculate_most_common_tag(self):
        movies_df = load_dataset(self.spark, self.movies_path)
        tags_df = load_dataset(self.spark, self.tags_path)
        result_df = calculate_most_common_tag(movies_df, tags_df)
        self.assertIsNotNone(result_df.head(1))

    def test_calculate_most_common_genre(self):
        ratings_df = load_dataset(self.spark, self.ratings_path)
        movies_df = load_dataset(self.spark, self.movies_path)
        result_df = calculate_most_common_genre(ratings_df, movies_df)
        self.assertIsNotNone(result_df.head(1))

if __name__ == "__main__":
    main()
    unittest.main(argv=['first-arg-is-ignored'], exit=False)