# ---
# jupyter:
#   jupytext:
#     text_representation:
#       extension: .py
#       format_name: light
#       format_version: '1.5'
#       jupytext_version: 1.14.4
#   kernelspec:
#     display_name: Python 3
#     name: python3
# ---

# + id="m3mD2VBa6AVy" colab_type="code" colab={}
# !apt-get install openjdk-8-jdk-headless -qq > /dev/null

# + id="seW_jDbB7uoK" colab_type="code" colab={}
# !wget -q www-us.apache.org/dist/spark/spark-2.4.3/spark-2.4.3-bin-hadoop2.7.tgz

# + id="hrJJXEQo70Kw" colab_type="code" colab={"base_uri": "https://localhost:8080/", "height": 1000} outputId="31324658-102f-44c9-9dad-194cf8036104"
# !tar -xvf spark-2.4.3-bin-hadoop2.7.tgz

# + id="-JZ2WuaF8JYp" colab_type="code" colab={}
# !pip install -q findspark

# + id="tWOG-a0g735U" colab_type="code" colab={}
import os
os.environ["JAVA_HOME"] = "/usr/lib/jvm/java-8-openjdk-amd64"
os.environ["SPARK_HOME"] = "/content/spark-2.4.3-bin-hadoop2.7"

# + id="vZNPSjeV78VD" colab_type="code" colab={}
import findspark
findspark.init()
from pyspark.sql import SparkSession
from pyspark.sql import Row
from pandas.plotting import scatter_matrix
import seaborn as sns
import numpy as np
from sklearn.linear_model import LinearRegression
from pyspark.sql.functions import array, col, explode, struct, lit

import warnings
warnings.simplefilter('ignore')

import pandas as pd
import matplotlib.pyplot as plt
# %matplotlib inline

spark = SparkSession.builder.master("local[*]").getOrCreate()

# + [markdown] id="_mMOn7Mp-EtY" colab_type="text"
# ## 7. Загрузить данные в Spark

# + id="7QdX2wYb8Dvu" colab_type="code" colab={}
df_data = spark.read.csv("u.data", sep='\t', header=None, inferSchema=True)
df_genre = spark.read.csv("u.genre", sep='|', header=None, inferSchema=True)
df_info = spark.read.csv("u.info", sep=' ', header=None, inferSchema=True)
df_occupation = spark.read.csv("u.occupation", sep=' ', header=None, inferSchema=True)
df_user = spark.read.csv("u.user", sep='|', header=None, inferSchema=True)
df_item = spark.read.csv("u.item", sep='|', header=None, inferSchema=True)
# encoding='latin_1'

# + id="hdlNr8Uc8aw7" colab_type="code" colab={}
# new_names_df_data = ['user_id', 'movie_id', 'rating', 'timestamp']
# df_data = df_data.toDF(*new_names_df_data)

# new_names_df_genre = ['genres', 'genres_id']
# df_genre = df_genre.toDF(*new_names_df_genre)

# new_names_df_user = ['user_id', 'age', 'gender', 'occupation', 'zip_code']
# df_user = df_user.toDF(*new_names_df_user)

# new_names_df_item = ['movie_id', 'movie_title', 'release_date', 
#                      'video_release_date', 'IMDb_URL', 'unknown', 
#                      'Action', 'Adventure', 'Animation', "Children's", 
#                      'Comedy', 'Crime', 'Documentary', 'Drama', 'Fantasy', 
#                      'Film-Noir', 'Horror', 'Musical', 'Mystery', 'Romance', 
#                      'Sci-Fi', 'Thriller', 'War', 'Western']
# df_item = df_item.toDF(*new_names_df_item)

# + id="3iVALWbYUzlg" colab_type="code" colab={"base_uri": "https://localhost:8080/", "height": 459} outputId="443df01e-092d-4d62-ab88-b738a5987844"
df_data.show()

# + id="RyKiE7SIU5Tn" colab_type="code" colab={"base_uri": "https://localhost:8080/", "height": 34} outputId="db9d3b1e-cf52-4c55-afba-20a06f0688da"
df_data.dtypes

# + id="8gN80j-BU55E" colab_type="code" colab={"base_uri": "https://localhost:8080/", "height": 187} outputId="0a29dceb-8126-4ea8-8354-8a044cedcc39"
df_data.describe().show()

# + [markdown] id="iiyzVgFmB4rz" colab_type="text"
# ## 8. Средствами спарка вывести среднюю оценку для каждого фильма

# + id="mibsuecQTs6x" colab_type="code" colab={"base_uri": "https://localhost:8080/", "height": 459} outputId="c51bbdc6-276c-4ce5-e1c3-f1825e373bae"
df_data.columns

df_data = df_data.withColumnRenamed('_c0', 'user id')\
    .withColumnRenamed('_c1', 'item id')\
    .withColumnRenamed('_c2', 'rating')\
    .withColumnRenamed('_c3', 'timestamp');

df_data_grp = df_data.groupby('item id')
df_data_grp_mean = df_data_grp.mean('rating')

df_data_grp_mean.show()

# + [markdown] id="9uOSL73oJO1K" colab_type="text"
# ## 9. В Spark получить 2 df с 5-ю самыми популярными и самыми непопулярными фильмами (по количеству оценок, либо по самой оценке)

# + id="sSst65eQEI5P" colab_type="code" outputId="5029e0aa-7fc4-48e5-b57e-98446776012a" colab={"base_uri": "https://localhost:8080/", "height": 357}
df_items = df_item['_c0', '_c1']

df_items = df_items.withColumnRenamed('_c0','item id')\
    .withColumnRenamed('_c1','movie title')

df_data_grp_mp = spark.createDataFrame(df_data_grp.count().orderBy('count', ascending=False).take(5)) 
df_data_grp_lp = spark.createDataFrame(df_data_grp.count().orderBy('count', ascending=True).take(5)) 

df_data_grp_mp.join(df_items, 'item id', how='inner').show()
df_data_grp_lp.join(df_items, 'item id', how='inner').show()

# + [markdown] id="Qr9MHlWHOQWS" colab_type="text"
# ## 10. Средствами Spark соедините информацию по фильмам и жанрам (u.genre)

# + id="pkLPdwhKE_GY" colab_type="code" outputId="48f5bbe1-ade3-4d9f-895c-7fe8b8a86460" colab={"base_uri": "https://localhost:8080/", "height": 867}
df_item.show()
df_genre.show()

# + id="StABwIswWPN5" colab_type="code" colab={}
df_item = df_item.withColumnRenamed('_c0','item id')\
            .withColumnRenamed('_c1','movie title')\
             .withColumnRenamed('_c5','unknown')\
        .withColumnRenamed('_c6','Action')\
        .withColumnRenamed('_c7','Adventure')\
        .withColumnRenamed('_c8','Animation')\
        .withColumnRenamed('_c9','Children\'s')\
        .withColumnRenamed('_c10','Comedy')\
        .withColumnRenamed('_c11','Crime')\
        .withColumnRenamed('_c12','Documentary')\
        .withColumnRenamed('_c13','Drama')\
        .withColumnRenamed('_c14','Fantasy')\
        .withColumnRenamed('_c15','Film-Noir')\
        .withColumnRenamed('_c16','Horror')\
        .withColumnRenamed('_c17','Musical')\
        .withColumnRenamed('_c18','Mystery')\
        .withColumnRenamed('_c19','Romance')\
        .withColumnRenamed('_c20','Sci-Fi')\
        .withColumnRenamed('_c21','Thriller')\
        .withColumnRenamed('_c22','War')\
        .withColumnRenamed('_c23','Western')\
    
df_item = df_item['item id', 'unknown', 'Action', 'Adventure', 'Animation',
              'Children\'s', 'Comedy', 'Crime', 'Documentary', 'Drama', 'Fantasy',
              'Film-Noir', 'Horror', 'Musical','Mystery','Romance', 'Sci-Fi',
              'Thriller', 'War','Western']


# + id="_ShBLmnXWPXJ" colab_type="code" colab={"base_uri": "https://localhost:8080/", "height": 459} outputId="9acd7cc4-422d-4fda-b811-f4bd498b45d7"
def to_long(df, by):

    cols, dtypes = zip(*((c, t) for (c, t) in df.dtypes if c not in by))
    kvs = explode(array([
      struct(lit(c).alias("key"), col(c).alias("val")) for c in cols
    ])).alias("kvs")

    return df.select(by + [kvs]).select(by + ["kvs.key", "kvs.val"])

df_g_trans = to_long(df_item, ["item id"])
df_g_trans = df_g_trans.where(df_g_trans['val'] > 0)

df_res = df_g_trans.join(df_items, 'item id', how='inner')['item id', 'movie title', 'key']\
    .withColumnRenamed('key','genre')
df_res = df_res.join(df_data_grp_mean, 'item id', how='inner') 
df_res.show()

# + [markdown] id="ucV6_JlKWk1S" colab_type="text"
# ## 11. Посчитайте средствами Spark среднюю оценку для каждого жанра

# + id="d45T7H6kWPed" colab_type="code" colab={"base_uri": "https://localhost:8080/", "height": 425} outputId="dddd283e-a6c1-46ff-8f30-6891d908473e"
df_data_grp_g = df_res.groupby('genre')
df_data_grp_mean_g = df_data_grp_g.mean('avg(rating)')
df_data_grp_mean_g.show()

# + id="GrJ73IhZNdqI" colab_type="code" colab={}
pass

StarcoderdataJupyterScriptsDedupFiltered