ComputeGraph Homework

computation graphs map reduce join merge request

Задание

1. Создайте git-ветку compgraph для решения задачи. Рекомендации:

   • Если делаете это впервые, посмотрите обязательно видео 8.3.GitBranches и почитайте секцию Как оформить merge request
   • Создайте ветку ДО ТОГО, как начали делать задание
   • Не забывайте переключаться обратно на master, когда будете коммитить другие задачи

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

   • Снабдите весь код docstring-ами, typing-aми

3. В папке unit_tests(папку memory_tests модифицировать не нужно) проверьте юнит-тестами.

   • все MapReduce-операции, на которые нет тестов в либе
   • все методы интерфейса графа
   • что последовательные запуски одного и того же графа не влияют друг на друга
   • code coverage должен быть больше 95%. Чтобы проверить локально см. pytest-cov В тестирующей системе используются следующие параметры
     pytest --cov-report=term-missing --cov=path/to/folder --cov-fail-under=95

4. Используя вашу библиотеку решите 4 задачи, приведенные ниже. Задачи должны целиком решаться в модели вычислительного графа - от чтения входа до записи результата и лежать в файле algorithms.py. На них затем запускаются тесты memory_tests/test_algorithms.py

5. Для каждой задачи сделайте скрипты, позволяющие быстро запустить её на входных данных из папки resources. Оформите это в папке examples. Отдельным бонусом будет визуализация для последней задачи.

6. Напишите Readme как установить, пользоваться библиотекой, запустить примеры и тесты. (Вариант темплейта Readme)
   Сам ридми нужно назвать LIB_README.md чтобы не перезатирать условие

7. Подкорректируйте setup.cfg своей метаинформацией

8. Отправьте Merge-Request, поставьте ему таг compgraph и проверьте корректность MR по чек-листу

9. Ждите проверки MR, баллы выставляются вручную после дедлайна задачи.

Важно! Нет тега -> нет проверки -> 0 баллов

Баллы

Базовый пакет

• 350 баллов и зачет ДЗ за принятый нами мердж-реквест (ревьювер проверяет реквест по чек-листу)
• +25 баллов бонуса, если с первого раза мердж-реквест будет корректным
• +10 баллов бонуса за визуализацию последней задачи (нужно указать в описании MR, что вы её сделали)

Note: с первого раза = в момент выставления тега, до этого мы не видим ваш MR вообще.

С ревью

• +50 баллов бонуса, если поправите ВСЕ замечания ревьювера (или аргументируете, почему какое-то замечание править не нужно)
• ставим 0 баллов и незачет за всю ДЗ, если вы в какой-то момент решите выйти из игры и не править замечания :) Это сделано так, потому что ревьювер тратит очень много времени на то, чтоб написать комментарии, и ему обидно, когда его игнорируют -_-

Важно! Чтоб участвовать в ревью - поставьте таг review на ваш мерджреквест.

У нас мало ревьюверов, поэтому устраиваем "Клуб 27". Если желающих на ревью будет много, то возьмем первые 27 реквестов по времени последнего апдейта. Кто раньше добавил последнее исправление - проходит.

Про процесс ревью

Если вы решили проходить ревью, то вашу работу с точки зрения стиля, дизайна, бестпрактис будет проверять ревьювер из нашей команды.
У вас будет 1-2-3 итерации, где вам будут присылать замечания на исправления. Замечаний может быть очень много. Замечания n итерации могут быть даже про код, который был написан ещё на первой.

Проверять начнем после наступления дедлайна. Исправлять можно до конца учебного года.

Во время ревью мы ставим таги:

• fix it - поправить код по комментариям
• reviewed - ревью завершено, оценка выставлена

Когда почините код, ставьте таги

• questions - если есть вопросы
• fixed - если можно смотреть правки. Снимайте наш таг fix it в этом случае

Не забывайте помечать комментарии выполненными (resolve), чтоб было проще ориентироваться (в правом верхнем углу комментария).

Чеклист содержимого мерджреквеста

• ✅ Стоит таг compgraph (иначе мы его не заметим даже)
• ✅ Находятся исключительно изменения в папке compgraph между версией из репозитория с задачами и вашей итоговой реализацией Проверяйте это по вкладке Changes
• ✅ Нет других задач, кроме compgraph
• ✅ Нет лишних случайно добавленных файлов
• ✅ Не добавили разархивированных ресурсов
• ✅ Прошли тесты в CI. Проверьте, что Pipeline в реквесте passed
• ✅ Реализованы Граф/Map/Reduce/Sort/Join
• ✅ Реализованы 4 задачи
• ✅ Есть скрипты на каждую задачу
• ✅ Есть Readme как пользоваться библиотекой и запускать задачи
• ✅ Есть тесты на все методы интерфейса графа
• ✅ Есть тесты на все операции, на которые нет тестов (все виды джойнов, мапперов, редьюсеров)
• ✅ Есть тесты на последовательные запуски графа

Как оформить merge request

Чтобы сдать задачу, нужно оформить правильный merge request, который содержит среди изменений код компграфа и только его.

Когда начинаем работу с компграфом:

• git pull upstream master - подтягиваем свежий master
• git checkout master - переходим обратно в мастер, если вдруг до этого были не там
• git branch compgraph - создаем ветку, в которой будем вести работу над compgraph
• git checkout compgraph - переходим в ветку с компграфом. Часть других заданий может пропасть - это нормально
• ...делаем все необходимые изменения...
• git add <измененные файлы>, git commit -m "meaningful commit message" - коммитим изменения
• git checkout master - переходим обратно в ветку с основными заданиями: изменения в компграфе пропадут, появятся новые задания, это нормально

Когда работа с компграфом закончена:

• git checkout compgraph - переходим в ветку, в которой велась работа над компграфом
• git push origin compgraph - заливаем последнюю версию на сервер. После этого в консоле появится ссылка на создание merge request, нужно по ней перейти
  Картинка

  
• убеждаемся в корректности реквеста по чек-листу
• проставляем лейбл compgraph (и review, если хотите его проходить) и нажимаем submit merge request (подробности на скрине)
  Картинка

  

Про задачу

В этом задании вы продолжите работать над библиотекой для вычислений над графами.

Напомним определение таблицы из предыдущей домашки:

Таблица - это последовательность словарей, где каждый словарь — это строка таблицы, а ключ словаря — колонка таблицы
(индекс в последовательности + ключ в словаре задают ячейку).

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

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

Для простоты можно считать, что все строки во входных таблицах содержат одинаковый набор ключей. При этом вашим операциям не запрещается создавать строки, у которых для некоторых ключей значения не определены, например: {'key1': 1, 'key2': None}, но в этом случае необходимо сделать реализацию всех операций устойчивой к такому поведению.

Зачем вообще нужны вычислительные графы?

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

Второй пример связан с уже знакомыми вам операциями Map / Reduce, и реализацией подобного графа мы и займемся с некоторыми оговорками:

• Будем выполнять код прямо на ваших компьютерах (это же учебное задание, в конце концов)
• Представим, что данные потенциально могут не влезать целиком в память (но все же их объем конечен)
• Будем обращать внимание на производительность нашего решения (об этом ниже)

Интерфейс графа вычислений

Граф вычислений состоит из точек входа для данных и операций над ними.

Вот так может выглядеть (хотя кого я пытаюсь обмануть, так и есть - см. compgraph/algorithms.py) граф, который подсчитывает кол-во слов в документах:

graph = Graph.graph_from_iter('texts') \
    .map(operations.FilterPunctuation('text')) \
    .map(operations.LowerCase('text')) \
    .map(operations.Split('text')) \
    .sort(['text']) \
    .reduce(operations.Count('count'), ['text']) \
    .sort(['count', 'text'])

Ещё раз заострим внимание: в момент создания графа мы не производим никаких чтений данных и вычислений.

Обратите внимание на интерфейс графа, каждая операция задается вызовом соответствующего метода класса Graph (полный список операций см. в compgraph/graph.py) - это рекомендуемая реализация, которая тем не менее не является обязательной. Вы можете менять интерфейс графа на своё усмотрение (единственное, что требуется - тесты должны работать как есть, без изменений).

Входные данные могут подаваться как в виде имен файлов с таблицами, которые необходимо открыть и прочитать, так и в виде произвольных генераторов, которые возвращают по одной строке за раз (так сделано в тестах, за примерами обращайтесь туда). Обратите внимание, что генераторы подаются на вход графу в виде фабрик, то есть в виде объектов, которые надо позвать, чтобы получить итератор. Это нужно для того, чтобы разные узлы графа вычислений могли читать вход несколько раз.

Также вы можете менять внутреннюю структуру библиотеки. Например, файл с операциями сейчас - это полная копия заготовки из предыдущего семинарского задания. Такая структура вполне может оказаться неудобной для вас, поэтому мы не стали завязываться в явном виде на ваш код прошлого задания, и вы можете как полностью скопировать его сюда, так и переписать.

Таблицы в файлах хранятся в виде последовательностей json-словарей (по одному на строку), каждый из которых описывает одну строку итоговой таблицы (см. директорию resources).

Таблицы, получаемые из генераторов нам показалось удобным реализовать так, чтобы в самом графе указывалось лишь ключевое слово для идентификации источника данных, тогда как сами данные передаются непосредственно во время запуска:

dummy_graph = Graph.graph_from_iter(name='docs')

graph.run(docs=lambda: iter([{'key': 'value'}]))

Создание графов над файлами мы не стали покрывать тестами, однако предоставляем вам набор данных, на которых вы должны самостоятельно протестировать ваш код (это обязательное условие сдачи, см. ниже).

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

Над входными данным мы будем запускать всё те же знакомые по предыдущему заданию операции. В общем случае операции вызываются последовательно, но есть операции (join), которые на вход принимают другие графы вычислений; за счет этого полный граф вычислений может быть нелинеен.

Очень важный момент: нелинейность в графе может возникнуть и без join'а:

graph = Graph().operation1(...).operation2(...)

# Caution! Non-linear execution flow
graph1 = graph.operation3()
graph2 = graph.operation4()

# Pure evil (he-he-he)
final_graph = graph1.join(..., graph2, ...)

Обработать такой момент в своём коде с сохранением потоковости исполнения --- сложно. Но при условии, что вы можете читать входные данные несколько раз, можно избавиться от необходимости делать узлы в графе вычислений с развилками.

После описания граф нужно уметь запускать, передав ему все нужные входные данные, каким-нибудь методом run:

def run(self, **kwargs) -> List[Row]:
    pass

Обратите внимание, что однажды созданный граф нужно уметь запускать на разных входах без пересоздания (и мы это проверяем в тестах).

Важный момент: как и в настоящем Map-Reduce, в нашем графе вычислений вам придётся сортировать данные. Отсортировать данные в потоковой манере невозможно, и это алгоритмически сложный примитив; поэтому в рамках нашего задания мы поступим хитрее. Операция сортировки за вас уже реализована в файле compgraph/external_sort.py путём делегирования сортировки отдельному процессу. Благодаря этому вы можете сконцентрироваться на том, чтобы ваше решение работало в стриминговой манере и не задумываться про потребление памяти на фазе сортировки. Можете ради интереса почитать устройство делегации стороннему процессу, это интересно.

Это очень похоже на то, как в настоящем Map-Reduce вы тоже зовёте готовый примитив Sort, и не задумываетесь, как он устроен.

Как запустить тесты?

Перед тем, как запустить тесты, нужно установить библиотеку.

# Устанавливаем библиотеку compgraph
$ ~/.pyenv/versions/shad_env/bin/pip install -e compgraph --force-reinstall

# Стал доступен модуль compgraph в интерпретаторе
# Теперь можете запустить тесты, которые используют модуль compgraph в импортах
$ ~/.pyenv/versions/shad_env/bin/pytest compgraph

Задачи для решения с помощью вашей библиотеки

Во всех задачах формат результата должен соответствовать тестам.

Word Count

Задача для разминки (базовый граф уже сделан, можно переделать по желанию).

В этой задаче вам дана таблица со строками в формате {'doc_id': ..., 'text :...'}. Требуется для каждого из слов, встречающихся в колонке text, посчитать количество вхождений во всю таблицу в сумме.

Файл с данными для этой и двух следующих задач: resource/text_corpus.txt

Инвертированный индекс на tf-idf

Работать будем с той же таблицей, что и в предыдущей задаче.

Для этой коллекции построим инвертированный индекс — структуру данных, которая для каждого слова хранит список документов, в котором оно встречается, отсортированный в порядке релевантности.

Релевантность будем считать по метрике tf-idf.

Для каждой пары (слово, документ) tf-idf зададим так:

TFIDF(word_i, doc_i) = (frequency of word_i in doc_i) * log((total number of docs) / (docs where word_i is present))

Для каждого слова надо посчитать топ-3 документов по tf-idf.

Когда будете делить текст на слова, не забудьте убрать пунктуацию (аналогично предыдущей задаче).

Подробный разбор задачи

На входе нам дана таблица с колонками doc_id, text.

На выходе для каждого слова из исходных документов хотим получить последовательность doc_id в порядке убывания tf-idf для этого слова в этом документе.

Будем решать поэтапно:

1. Создадим промежуточный граф, который пропустит вход нашей исходной таблички через mapper, который разделит каждую строку с текстом на слова.

   split_word := input('docs') -> map(split_words)
   

2. Создадим ещё один граф, который посчитает количество документов в таблице (оно нужно нам для формулы idf). Он будет состоять из единственной операции reduce, которая будет считать количество строчек в переданной ей таблице.

   count_docs := input('docs') -> reduce(count_rows)
   

3. Теперь опишем граф, считающий idf каждого слова. Входом для этого графа будет выход графа из п.1. Для начала запустим на этой таблице reduce по ключу ('doc_id', 'word') и для каждого ключа оставим только одно его вхождение (ведь нас интересует количество документов в которых встретилось слово, без разницы сколько раз). Результат этой операции отсортируем и будем редьюсить уже по словам (т. е. по word), считая количество документов, в которых это слово встретилось. Далее сджойним граф из п.2 с нашей таблицей, используя inner join, фактически дописав в каждую строку таблицы общее кол-во текстов. Полученную таблицу прогонним через маппер, который посчитает для каждой строки idf, используя имеющуюся информацию о кол-ве документов, содержащих слово, и суммарном кол-ве документов.

   count_idf := input(split_words) -> sort('doc_id', 'word') ->
        -> reduce(first, keys=('doc_id', 'word')) -> sort('word')
        -> reduce(count, keys=('word')) ->
        -> join(inner, count_docs) -> map(idf)
   

4. Создадим ещё граф. Входом для него так же, как для предыдущего графа, будет результат графа из п.1, но в этот раз мы его будем редьюсить по doc_id, считая частоту каждого слова, то есть числитель нужной нам формулы.

   tf := input(split_word) -> sort('doc_id') -> reduce(tf, 'doc_id')
   

5. После этого нужно сделать join c результатом графа из п.3 и перемножить соответствующие метрики. Последним будет reduce по word, который выберет для каждого слова top-3 документа по tf-idf.

Топ слов с наибольшей взаимной информацией

Задача, обратная предыдущей: для каждого документа посчитать топ-10 слов, наиболее характерных для него.

Ранжировать слова будем по метрике Pointwise mutual information.

Более формально задача ставится так: для каждого текста надо найти топ-10 слов, каждое из которых длиннее четырех символов и встречается в каждом из документов не менее двух раз; топ надо выбирать по величине:

pmi(word_i, doc_i) = log((frequency of word_i in doc_i) / (frequency of word_i in all documents combined))

Upd. Если у вас не сходятся результаты расчетов с тестами, то обратите внимание, что слова, не соответствующие условию выкидываются ещё до расчетов частот. Воспринимайте это как "фичу" :)

Средняя скорость движения по городу от часа и дня недели

В этой задаче вам надо работать с информацией о движении людей на машинах по какому-то подмножеству улиц города Москвы.

Улицы города заданы как граф, а информация о передвижении задана как таблица, в каждой строке которой данные вида

{'edge_id': '624', 'enter_time': '20170912T123410.1794', 'leave_time': '20170912T123412.68'}

где edge_id — идентификатор ребра графа дорог (то есть просто участка какой-то улицы), а enter_time и leave_time — соответственно время въезда и выезда с/на это ребро (время в UTC).

Также вам дана вспомогательная таблица вида

{'edge_id': '313', 'length':121, 'start': [37.31245, 51.256734], 'end': [37.31245, 51.256734]}

где length - длина в метрах, start и end — координаты начала и конца ребра, заданные в формате ('lon', 'lat'). Быть может, не для всех рёбер графа есть вся метаинформация.

Пользуясь этой информацией вам надо построить таблицу со средней скоростью движения по городу в км/ч в зависимости от часа и дня недели:

{'weekday': 'Mon', 'hour': 4, 'speed': 44.812}

Для проверки полезно построить график по этой таблице, он должен выглядеть предсказуемо (если покажете нам, дадим бонусные баллы).

Файлы для этой задачи: resource/travel_times.txt и resource/road_graph_data.txt