Задачи Apache Spark

##Задача 1 Модернизируйте заготовку заменив все участки на ваш код для того, что бы:

вычислять и выводить на экран статистику по числу оценок для каждого фильма

вычислять и выводить на экран статистику по числу оценок для всех фильмов

Статистика для каждого фильма:

Marks for film 346: 1 -> 7, 2 -> 10, 3 -> 32, 4 -> 49, 5 -> 28

Marks for film 474: 1 -> 0, 2 -> 6, 3 -> 34, 4 -> 59, 5 -> 95

Marks for film 265: 1 -> 1, 2 -> 13, 3 -> 62, 4 -> 91, 5 -> 60

Marks for film 465: 1 -> 4, 2 -> 8, 3 -> 26, 4 -> 30, 5 -> 17

Marks for film 451: 1 -> 15, 2 -> 31, 3 -> 37, 4 -> 54, 5 -> 33

Marks for film 86: 1 -> 4, 2 -> 10, 3 -> 23, 4 -> 67, 5 -> 46

Marks for film 257: 1 -> 2, 2 -> 28, 3 -> 81, 4 -> 126, 5 -> 66

Marks for film 222: 1 -> 7, 2 -> 30, 3 -> 108, 4 -> 155, 5 -> 65

Marks for film 40: 1 -> 9, 2 -> 9, 3 -> 20, 4 -> 17, 5 -> 2

Marks for film 29: 1 -> 15, 2 -> 34, 3 -> 45, 4 -> 14, 5 -> 6

Для всех фильмов:

Marks for films ALL: 1 -> 6110, 2 -> 34174, 3 -> 27145, 4 -> 11370, 5 -> 21201

import collections

rdd = spark.sparkContext.textFile("/content/sample_data/ml-100k/u.data")

#

def printStat(inp):

#

print(f'Marks for film {ind}: 1 -> {marks[0]}, 2 -> {marks[1]}, 3 -> {marks[2]}, 4 -> {marks[3]}, 5 -> {marks[4]}')

for i in aggPairRDD.mapValues(lambda x: dict(collections.Counter(x))).collect():

printStat(i)

#

##Задача 2 Произведите подсчёт частоты встречаемости слов с использованием ApacheSpark RDD. Ячейка ниже скачивает текст. Вам требуется:

Очистить текст от знаков препинания и пустых строк

Перевести в нижний регистр и разделить по пробелам

Подсчитать наиболее часто встречающиеся символы

Использовать RDD

Пример вывода:

[('и', 2204),

('в', 1977),

('я', 1252),

('не', 1247),

('на', 1094),

('он', 755),

('как', 717),

('с', 693),

('что', 653),

('его', 502)]

!wget http://www.lib.ru/INOOLD/BALZAK/shagren.txt_Ascii.txt | iconv -f cp1251

i = 0

with open('/content/shagren.txt_Ascii.txt', encoding="cp1251") as inF, open('/content/shagren.txt_utf8.txt', "w") as outF:

for line in inF:

outF.write(line)