Реализация коллаборативной фильтрации при помощи Hadoop Streaming и Python

Запуск Hadoop-кластера

Hadoop-кластер запускался при помощи сети докер контейнеров, доступных по ссылке https://github.com/nakhodnov17/docker-hadoop-spark.

Сам скрипт запускается следующей последовательностью команд:

• docker cp dir_name namenode:/
• docker exec -it namenode bash
• cd dir_name
• sh run

Neighborhood подход в коллаборативной фильтрации

Имея матрицу user-item из оценок пользователей можно определить меру adjusted cosine similarity похожести товаров $i$ и $j$ как векторов в пространстве пользователей:

\begin{equation} sim(i, j) = \frac{\sum_{u \in U_{i,j} (r_{u,i} − /bar{r_u})(r_{u,j} − /bar{r_u})}{\sum_{u \in U_{i,j}(r_{u,i} − /bar{r_u})^2\sum_{u \in U_{i,j}(r_{u,j} − /bar{r_u})^2}, \end{equation}

где $U_{i,j}$ – множество пользователей, которые оценили фильмы $i$ и $j$, $\bar{r_u}$ – средний рейтинг пользователя $u$. Рейтинги для неизвестных фильмов считаются по следующей формуле:

\begin{equation} \hat{r_{u,i}} = \frac{\sum_{j \in I_u} sim(i,j) r_{u,j}}{\sum_{j \in I_u}sim(i,j)}, \end{equation}

где $I_u$ - множество фильмов, которые оценил пользователь $u$. Такой подход называется item-oriented. Обратим внимание на то, что $sim(i, j) \in [-1, 1]$. Это может привести к делению на ноль или значениям $\hat{r_{u,i}}$ вне диапазона [0, 1]. Избавиться от этой проблемы можно, например, положив равными нулю отрицательные значения $sim(i, j)$.

Описание задачи

Необходимо реализовать коллаборативную фильтрацию с использованием фреймворка MapReduce. Ваша программа, получая на вход список троек (пользователь, фильм, рейтинг) и список соответствий между номером фильма и его названием, должна вывести для каждого пользователя топ-100 фильмов с самым высоким предсказанным рейтингом.

При вычислений рекомендаций необходимо учитывать только те фильмы, которые пользователь ещё не оценил. Рекомендации выводятся по убыванию предсказанной оценки. При равенстве предсказанных оценок выше в списке рекомендаций должен стоять фильм с лексикографически меньшим названием.

Файл с предсказаниями необходимо представить в следующем виде:

<user_id>@<rating_1>#<film_name_1>@...@<rating_100>#<film_name_100>

В качестве датасета предлагается использовать «small» версию датасета MovieLens (https://grouplens.org/datasets/movielens/latest/).

Логика алгоритма

Алгоритм был реализован с помощью парадигмы MapReduce. Он был разбит на 4 последовательно выполняемых MapReduce-задачи, общая схема работы представлена ниже.

                                          input
                                            |
                                            \/
                                     ------------
                                    |   Step 1   |
                                     ------------
                                            |
                               -------------|
                              |             |
                              |             \/
                              |      ------------
                              |     |   Step 2   |
                              |      ------------
                              |            |
                              \/           \/
                             ----------------------
                            |        Step 3        |
                             ----------------------
                                    |
                                    |
                                    |
                                    \/
                                 ------------
                                |   Step 4   |
                                 ------------
                                    |
                                    \/
                                  final

Step 1

Чтение данных

Mapper просто парсит данные таблиц movies.csv и ratings.csv. Отбрасываются заголовки столбцов, данные из разных таблиц снабжаются соответствующими метками и приводятся к внутреннему представлению. Mapper не использует дополнительную память, за время работы он выполняет $\frac{2UI\alpha}{M}$ операций сложения и $\frac{2(UI\alpha + I)}{M}$ операций сравнения.

Reducer агрегирует записи таблицы ratings.csv в записи вида пользователь-список оценённых фильмов, а записи таблицы movies.csv передаёт на следующий этап без изменений. Этот процесс также не использует дополнительную память, а количество выполненных им операций равно $\frac{2(UI\alpha + I)}{R}$ (сравнений) и $\frac{UI\alpha}{R}$ (сложение).

Step 2

Построение матрицы сходства

Mapper также передаёт записи таблицы movies.csv на следующий этап. На основе данных из таблицы ratings.csv он генерирует всевозможные попарные сочетания оценок фильмов из базы одним пользователем, кроме того, вместе с этими записями он сохраняет и средний рейтинг каждого пользователя. В памяти процесс хранит $\alpha I$ фильмов, оценённых одним пользователем, за время работы он выполняет $\frac{UI\alpha}{M}$ операций сложения, и $\frac{U}{M}$ операций сравнения и $\frac{U}{M}$ операций деления.

Reducer вычисляет элементы матрицы сходства согласно формуле $(1)$ из условия и добавляет к этим записям название первого фильма из пары на основе данных из таблицы movies.csv. Его затраты по памяти минимальны, а общее число арифметических и логических операций равно $\frac{I^2(15 U \alpha^2 + 2)}{R}$.

Step 3

Предсказание индивидуального рейтинга

Mapper оставляет без изменений данные, полученные на предыдущем этапе и преобразует данные первого этапа, создавая записи вида пользователь-непросмотренный им фильм. В памяти процесс хранит $\frac{I(U\alpha + 1)}{M}$ элементов (индексы фильмов), за время работы он выполняет $\frac{U + I^2}{M}$ операций сравнения.

Reducer предполагаемые рейтинги фильмов для разных пользователей согласно формуле $(2)$ из условия, также он добавляет к записям названия этих фильмов из таблицы movies.csv. Этот процесс хранит в памяти $\frac{I(U\alpha + 1)}{R}$ объектов, а общее число арифметических и логических операций равно $\frac{I(I + (1 - \alpha) U + 5))}{R}$.

Step 4

Формирование рекомендаций

Mapper слегка меняет внутреннее представление строк и готовит их для сортировки перед подачей редьюсеру. В памяти он ничего не хранит, за время работы выполняет $\frac{2(1 - \alpha)IU}{M}$ операций сложения.

Reducer агрегирует полученные оценки фильмов, сопоставляя каждому пользователю $100$ новых для него фильмов с наивысшими предполагаемыми рейтингами. Рекомендации получены! Этот процесс хранит в памяти $100$ объектов (топ-100 фильмов), и выполняет $\frac{(1 - \alpha)I U }{R}$ операций сравнения и $\frac{3(1 - \alpha)I U + 200}{R}$ операций сложения.

Общее время работы алгоритма составило 20 минут 50 секунд

CPU: Intel Core i5-6500 3.20GHz

RAM: 16 Gb DDR3