Пост 12. Плоский файл со смещением.

Доброго времени суток!

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

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

Например, есть файл, в котором имеются 4 колонки: первая это символы с 1 по 4, вторая с 5 по 10, третья с 11по 16, четвертая с 17 по 24.Как то так:

col1fieldfield2field#3

а хочется получить структуру вида:

col1       field      field2    field#3

Нет проблем – описываем следующую структуру:

create external table fixed_offset (a string,b string, c string,d string)

ROW FORMAT SERDE 'org.apache.hadoop.hive.contrib.serde2.RegexSerDe'

 WITH SERDEPROPERTIES

 (

 "input.regex" = "(.{4})(.{5})(.{6})(.{7}).*",

 "output.format.string" = "%1$s %2$s %3$s %4$s"

 )

 STORED AS TEXTFILE; 

И обращаемся через обычный select:

Select * from fixed_offset

Пост 11. Запись файлов в HDFS.

Доброго времени суток. Подумалось мне, что о самом главном я еще и не рассказывал. А точнее о том как файлы записываются в HDFS. Иллюстрирует это действо картинка ниже.

Клиент, инициируя запись файла обращается к NameNode (NN) и сообщает в какую директорию HDFS и он хочет записать файл. NN проверяет наличие прав у клиента, факт существования файла и.т.п. Если все хорошо – передает список DN на которые надо записать файл. Клиент  «пилит» файл на своей стороне и по блокам начинает передавать его в кластер, где внутри кластера по интерконнекту происходит репликация. Затем это действо повторяется со следующим блоком и так далее пока не будут записаны все блоки. Рядом с каждым блоком пишется файл контрольной суммы, который каждый раз сверяется с расчетной при чтении блока.

Запись блоков происходит последовательно. Как говорил Сталин Черчилю на ялтинской конференции: “доверяй, но проверяй” – давайте проверим. 

Файл из прошлого поста (он у меня около 4.2 Мб) загружаем с разными размерами блока.

Сначала 1 Мб (это 5 блоков).

hadoop fs -D dfs.blocksize=1048576 -put /tmp/alpha.txt /tmp/

загрузка произошла практически моментально.

Теперь попробуем записать с блоком 1 Кб (я не сошел с ума – просто экспериментирую).

hadoop fs -D dfs.blocksize=1024 -put /tmp/alpha.txt /tmp/

Время загрузки значительно увеличилось (идет запись 4400 блоков вместо 5), что позволило нам не спеша открыть рядышком вторую консольку и командой

netstat -na|grep 50010|grep EST

Воочию убедиться что открыто не 4400 сессий с DataNodes (порт 50010), а периодически открываются разные сессии. Кстати, эта особенность framework - еще одно объяснение почему Hadoop любит большие блоки.

Пост 10. Сортировка в hadoop. Синтаксис hive.

Итак, сегодня хотел бы рассказать об основных алгоритмах MapReduce и как они могут быть имплементированы в рамках данной парадигмы.

Сортировка – идея проста: на шаге map идет простая разметка, например на вход подается

(K0,V0) – (0, word1)

На выходы это преобразуется в

(K1,V1) – (word1,1)

Стоит заметить, что 0 во входной паре и 1 в выходной  - удовлетворение требования.

Далее идет стадия partition – на которой определяется на какой редьюсер пойдет выходная пара (если редьюсер один – то все становится просто)

                Вся магия сортировки в Hadoop – выбираем функцию partitioning такую, что если k1<k2, то и partitioning(k1) <= partitioning(k2).

Reduce не делает ничего. Framework уже все отсортироавл, прежде чем подавать это на reduce.

Давайте рассмотрим пример.

Создаем файл 12345 куда вбиваем алфавит:

A

B

C

….

Размножаем его:

for i in `seq 1 100000`; do cat 12345 >> /tmp/alpha.txt; done;

Загружаем в HDFS с размером блока 1 Мб

hadoop fs -D dfs.blocksize=1048576 -put /tmp/alpha.txt /tmp/

Проверяем количество блоков:

hadoop fsck /tmp/alpha.txt -locations -blocks –files

Создаем в HIVE  табличку с одной колонкой (я назвал ее sort_test)

Пишем запрос:

create table sort_result as

select col_0 from sort_test

sort by col_0

До того как его запустить выставим параметр

mapred.max.split.size=1048576

Иначе не будет нескольких map  - все свалится в один (с этим параметром у меня получилось 5 map), количество reduce тоже выставляем равным mapred.reduce.tasks=5

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

После того как запрос отработает можно посмотреть что получилось:

[oracle@localhost ~]$ hadoop fs -ls /user/beeswax/warehouse/sort_result/

Found 5 items

-rw-r--r--   1 oracle hadoop     878714 2013-03-24 12:46 /user/beeswax/warehouse/sort_result/000000_0

-rw-r--r--   1 oracle hadoop     879928 2013-03-24 12:46 /user/beeswax/warehouse/sort_result/000001_0

-rw-r--r--   1 oracle hadoop     878716 2013-03-24 12:46 /user/beeswax/warehouse/sort_result/000002_0

-rw-r--r--   1 oracle hadoop     879966 2013-03-24 12:46 /user/beeswax/warehouse/sort_result/000003_0

-rw-r--r--   1 oracle hadoop     882676 2013-03-24 12:47 /user/beeswax/warehouse/sort_result/000004_0

“Вытянем” это на локальную машину

hadoop fs -get /user/beeswax/warehouse/sort_result/*

И увидим 5 практически идентичных файлов.

Ну вот как то так оно все сортируется. Из hadoop все это можно вытащить в виде одного файла

hadoop fs -getmerge /user/beeswax/warehouse/sort_result/* big_file

но не стоит обольщаться – вы получите один большой файл вида:

a

…

z

a

…

z

…

То есть финальной сортировки не будет.

Если все таки хочется получить на выходе один файл в отсортированном порядке – можно указать вместо sort by конструкцию order by

create table sort_result as

select col_0 from sort_test

order by col_0

Пост 9. Hive и MapReduce. Агрегаты.

Итак, если вы внимательно прочитали все мои предыдцщие посты - вы теперь знаете достаточно, что бы воспринять Hive как MapReduce инструмент. Только задания Map и Reduce вы пишете не на Java, а на HQL (язык весьма напоминающий SQL, но там нет некоторых конструкций - одиночной вставки, update...). На самом деле MapReduce вы "пишете в двух метах" - при создании таблицы и при написании select.
Рассмотрим пример. У Вас есть плоский файл (csv), который лежит в HDFS. Плоский файл вида:
col1, col2, col3, col4
........
Лежит допустим где то в /user/examples/csv (естественно в hdfs).
Описываем его следующим DDL:
CREATE EXTERNAL TABLE `simplest_csv`
(
  `a` string ,
  `b` string ,
  `c` string ,
  `d` string )
ROW FORMAT   DELIMITED
    FIELDS TERMINATED BY ','
  STORED AS TextFile LOCATION "/user/examples/csv"

Возьмен конкретный пример файла, который поместим в HDFS:
333, wqer, 56, 88m
2, wqer, 56, 88m
334, wqert, 54, 88w
335, wqerty, 59, 88e

Напишем запрос к этиму файлу

select sum(a),b from simplest_csv

group by b

Что будет происходить на фазах Map и Reduce?

На вход будет подаваться строка, RR будет преобразовывать ее в пару ключ-значение (ключом будет являться смещением строки относительно начала файла). Внутри Map будет происходить преобразование к новой паре ключ-значение: будут "викинуты" не перечисленые колонки (так что перечисляйте только то что надо перечислить!) колонка (колонки) по которой идет группировка - будет ключом, та котонка которую надо суммировать - значением. Новая пара KV идет на Reduce (как и многие другие). Но перед пересылкой на Reduce в Hive всегда на агрегационных функцияю применяется combiner. Что делает reduce думаю объяснять не надо.

Описав этот процесс думаю так же бессмысленно говорить о том что подобные запросы линейно масштабируемы - я думаю все и так все поняли:)

Окей, с csv все вроде бы ясно. А как быть с полуструктурированными даными? Допустим с JSON? Все опять предельно просто и алгоритмично! 

Возьмем файл данных 

{column1:333, column2: wqer, column3: 56}
{column1: 2, column2: wqer, column3: 56}
{column1: 334, column2: wqert, column3: 54}
{column1: 335, column2: wqerty, column3: 59}

и положим его в HDFS

[root@cdh ~]# sudo -u hdfs hadoop fs -mkdir /user/examples/json

[root@cdh ~]# sudo -u hdfs hadoop fs -put /tmp/1234 /user/examples/json/

Создаем таблицу:

CREATE EXTERNAL TABLE simplest_json

(column1 string,column2 string,column3 string)

ROW FORMAT SERDE 'org.apache.hadoop.hive.contrib.serde2.JsonSerde'

LOCATION '/user/examples/json/'

И пишем запрос. Запрос к слову ничем не отличается от предыдущего:

select sum(column1),column2 from simplest_json

group by column2

И что же происходит? Практически то же самое что и в случае CSV, только на стадии Map происходит еще и парсинг строки (разбор JSON).

Просто магия какая то:)

Подытожим: 

1) Map:
- Перекомбинирование пар Key-Value
- Парсинг строчки
- Фильтрация по условию where
- Combiner (локальный reduce)
- Применение функций к некоторым
- Локальная свертка по колонкам если это необходимо

2) Reduce:
- свертка
Вывоы:

- Перечисляйте только нужные колонки (сократит выход map)
- Используйте где это возможно условие where (сократит выход map)

Пост 8. MapReduce для взрослых. Performance tuning.

Я продолжаю описывать принцип работы MapReduce заданий. Еще раз напоминаю, что hive - это целиком и полнстью MapReduce, просто описываете вы его языком более верхнего уровня, нежели java. Давайте посмотрем более детально, что же происходит, когда мы вновь пытаемся посчитать слова в "Евгение Онегине".
Как было сказано раньше, блоки бьются строго по размеру, никакой дополнительной логики нет.
Допустим строка "Когда не в шутку занемог" была разбита на 2 части, которые ушли в соответствующие блоки. В первый блок: "Когда не в шутку занем", а "ог" во второй.


Определив размеры входных порций (размер блока или параметр mapred.max.split.size)

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

Дальше начинается процесс Map схема изображенная выше - слишком поверхностная, давайте рассмотрим более подробную:

1)Map пишет выход в некий буфер (io.sort.mb по умолчанию 100 Мб). На вход получает (k0, v0) на выходе (k1, v1) - пишутся в этот самый буфер.
2) Когда буфер заполнен на io.sort.spill.percent (по умолчанию 0.8) в бэкграунде начинает происходить spill (сброс данных на диск), при этом Map продолжает писать в буфер, НО если буфер будет заполнен полностью – map заморозит работу до конца spill. Spill пишет на локальный диск (mapred.local.dir). Во время spill происходит:

- Разделение на reduce порции (partitioner) и запись в отдельные файлы (partition file).
- Сортировка по ключу каждой reduce порции.
- Опционально. Combiner – “локальный reduce”.
Если параметр mapred.compress.map.output = true, на диск пишутся компрессированные данные. Компрессионный кодек:mapred.map.output.compression.codec.

3) После того как map записывает последний spill, начинается объединение и сортировка каждого partition файла в один (в io.sort.factor потоков, по умолчанию один). Идеальный вариант: один map – один spill. Если partition файлов больше чем min.num.spills.for.combine (по умолчанию 3) – запускается еще раз combiner.

4) После окончания map стартует copy фаза reduce - начинает забирать свою порцию данных в mapred.reduce.parallel.copies потоков (по умолчанию 5 на каждый reduce) в JVM память reduce. Это т буфер регулируется параметром mapred.job.shuffle.input.buffer.percent

5) После того как отработал последний map начинается сортировка-слияние map-ов в один файл. Количество файлов (которое определяется количеством map) участвующих в одной итерации равно io.sort.factor (10 по умолчанию).

Например было 40 map, тогда sort phase выполнится в 5 итераций (4 по 10 + одна – финальная сортировка-слияние). В идеале количество map должно стремиться к io.sort.factor. Попутно происходит преобразование пар (K1,V1) в (K1,(ist(V))

6) Reduce программа написанная пользователем. Преобразование (K1,V1) в (K1,(ist(V)) в (K2,V2)

Все вышеописанное позволяет понять способы разгона и оптимизации MapReduce заданий. Я намерено не консолидировал их в таблицу стандартных советов, что бы читатель смог подумать, поэксперементировать, самостоятельно сделать выводы о значениях некоторых параметров.

Пост 7. Map-only задания.

В данном посте хотел бы рассказать о таком явлении как map only задании. Начав изучать hadoop и MapReduce я все больше больше задумывался о конкретной реализации различных операций в рамках MapReduce парадигмы, пытался представить что же будет делать шаг Map, а  что Reduce. Задача перевода текста в верхний региср. Хорошо, на стадии Map, все вроде бы понятно, берем кажрое слово по аналогии с wordcount и возвращаем его в верхнем регистре. А что делает reduce? Ничего. Есть такие задания, где выход Map задания и есть конечный ответ. Reduce просто напросто нет.

Примером таких Map-only MarReduce job могут быть:
- Обработка изображений
- ETL (sqoop, например)
- Конвертация из одного формата в другой

Мир hadoop удивителен:)
Если у вас есть какие либо вопросы - не стесняйтесь их задавать!

Пост 6. Combiner. Или как уменьшить колличество пересылок по сети.

Наверное каждый, кто начинал разбираться с hadoop и начинал с примера подсчета слов, задавал себе и окружающим один и тот же вопрос: зачем пересылать втупую все пары по сети, когда более эфективно было бы сделать локальную свертку, а только потом переслать по сети агрегаты. Hadoop не дураки писали, соответственно такая возможность есть!
Называется она combiner.

Как видно на картинке выше с выхода map пары key - value попадают на combiner, где выполняется код reduce. В итоге процент пересылки по сети значительно сокращается. Ура!  Но как вы пониимаете далеко не всегда можно использоавть данную возможность. В контексте задач вычисления агрегатов - "золотая" возможность, о которой вы должны помнить.

Если есть вопросы - не стесняйтесь их задавать!