Работа с заданиями MapReduce

MapReduce — инструмент параллельной обработки больших (порядка нескольких десятков ТБ) наборов данных на кластерах в экосистеме Hadoop. Позволяет работать с данными в разных форматах. Ввод и вывод задания хранится в Yandex Object Storage. MapReduce использует ряд библиотек, путь к которым определяется сборщиком Apache Bigtop.

В этой статье на простом примере показывается, как в Yandex Data Processing использовать MapReduce. При помощи MapReduce подсчитывается количество жителей 500 самых населенных городов мира из набора данных о городах.

Чтобы запустить MapReduce на Hadoop, используется интерфейс Streaming. При этом для этапов предобработки данных (map) и вычисления финальных данных (reduce) используются программы, читающие из стандартного программного ввода (stdin) и выдающие результат в стандартный вывод (stdout).

Перед началом работыПеред началом работы

1. Создайте сервисный аккаунт с ролями dataproc.agent и dataproc.provisioner.

2. В Object Storage создайте бакеты и настройте доступ к ним:

   1. Создайте бакет для исходных данных и предоставьте сервисному аккаунту кластера разрешение READ для этого бакета.
   2. Создайте бакет для результатов обработки и предоставьте сервисному аккаунту кластера разрешение READ и WRITE для этого бакета.

3. Создайте кластер Yandex Data Processing со следующими настройками:

   • Окружение — PRODUCTION.
   • Сервисы:
     • HDFS
     • MAPREDUCE
     • YARN
   • Сервисный аккаунт: выберите созданный ранее сервисный аккаунт.
   • Имя бакета: выберите бакет для результатов обработки.

Создайте задание MapReduceСоздайте задание MapReduce

1. Скачайте и загрузите в бакет для исходных данных архив CSV-файла с набором данных о городах.

2. Загрузите в бакет для исходных данных файлы на языке Python с кодом программ предобработки данных (этап map) mapper.py и вычисления финальных данных (этап reduce) reducer.py:

   mapper.py

   #!/usr/bin/python
   import sys
   
   population = sum(int(line.split('\t')[14]) for line in sys.stdin)
   print(population)
   

   reducer.py

   #!/usr/bin/python
   import sys
   
   population = sum(int(value) for value in sys.stdin)
   print(population)
   

3. Создайте задание MapReduce с параметрами:

   • Основной класс: org.apache.hadoop.streaming.HadoopStreaming
   • Аргументы:
     • -mapper
     • mapper.py
     • -reducer
     • reducer.py
     • -numReduceTasks
     • 1
     • -input
     • s3a://<имя_бакета_для_исходных_данных>/cities500.txt
     • -output
     • s3a://<имя_бакета_для_результатов_обработки>/<папка_для_результатов>
   • Файлы:
     • s3a://<имя_бакета_для_исходных_данных>/mapper.py
     • s3a://<имя_бакета_для_исходных_данных>/reducer.py
   • Настройки:
     • mapreduce.job.maps: 6
     • yarn.app.mapreduce.am.resource.mb: 2048
     • yarn.app.mapreduce.am.command-opts: -Xmx2048m

4. Подождите, пока статус задания изменится на Done.

5. Скачайте из бакета и просмотрите файл с результатом обработки:

   part-00000

   3157107417
   

Удалите созданные ресурсыУдалите созданные ресурсы

Некоторые ресурсы платные. Чтобы за них не списывалась плата, удалите ресурсы, которые вы больше не будете использовать:

1. Удалите кластер.
2. Удалите бакеты.
3. Удалите сервисный аккаунт.