Производительность и эффективность

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

Как включить Tez

Чтобы запустить Pig в режиме Tez, просто добавьте "-x tez" в командной строке Pig. В качестве альтернативы, вы можете добавить "exectype=tez" в conf/pig.properties, чтобы изменить тип выполнения по умолчанию на Tez. Системная переменная Java "-Dexectype=tez" также подходит для запуска режима Tez.

Предварительные условия: Tez требует наличия пакета tez в hdfs во время выполнения задачи на кластере и файла tez-site.xml с настройкой tez.lib.uris, указывающей на это местоположение в hdfs в пути к классам. Скопируйте пакет tez в hdfs и добавьте директорию конфигурации Tez ($TEZ_HOME/conf), содержащую tez-site.xml, в переменную среды "PIG_CLASSPATH", если pig в режиме tez завершается ошибкой с сообщением "tez.lib.uris is not defined". Это требуется дистрибутивом Apache Pig.

  <property>
    <name>tez.lib.uris</name>
    <value>${fs.default.name}/apps/tez/tez-0.5.2.tar.gz</value>
  </property>

Генерация DAG Tez

Каждый скрипт Pig будет скомпилирован в 1 или более DAG Tez (обычно 1). Каждый DAG Tez состоит из ряда вершин и рёбер, соединяющих вершины. Например, простое объединение включает 1 DAG, состоящий из 3 вершин: загрузка левого входного файла, загрузка правого входного файла и объединение. Выполнение explain в режиме Tez покажет вам DAG, скомпилированный из скрипта Pig.

Переиспользование сеансов/контейнеров Tez

Один из недостатков MapReduce — высокая стоимость запуска задачи. Это негативно сказывается на производительности, особенно для небольших задач. Tez решает эту проблему с помощью переиспользования сеансов и контейнеров, поэтому нет необходимости запускать master-приложение для каждой задачи и запускать JVM для каждой задачи. По умолчанию переиспользование сеансов/контейнеров включено, и обычно его не нужно отключать. Переиспользование JVM может привести к побочным эффектам, если используются статические переменные, так как статические переменные могут сохраняться между разными задачами. Поэтому, если статические переменные используются в EvalFunc/LoadFunc/StoreFunc, убедитесь, что реализована функция очистки и она зарегистрирована с помощью JVMReuseManager.

Автоматическое распараллеливание

Как и в MapReduce, если пользователь указывает «parallel» в своем заявлении Pig или определяет default_parallel в режиме Tez, Pig учтёт это (единственное исключение — если пользователь указывает явно слишком низкое значение параллелизма, Pig переопределит его).

Если пользователь не указывает ни «parallel», ни «default_parallel», Pig будет использовать автоматическое распараллеливание. В MapReduce Pig отправляет одну задачу MapReduce за раз, и перед отправкой задачи Pig имеет возможность автоматически установить параллелизм редьюсеров на основе размера входного файла. В отличие от этого, Tez отправляет DAG как единицу, и автоматическое распараллеливание выполняется в три этапа:

• Перед отправкой DAG Pig статически оценивает параллелизм каждой вершины на основе размера входного файла DAG и сложности конвейера каждой вершины.
• Когда DAG выполняется, Pig корректирует параллелизм вершин, используя лучшие доступные знания в данный момент (управление параллелизмом Pig).
• Во время выполнения Tez динамически корректирует параллелизм вершин на основе объема входных данных вершины. В настоящее время Tez может только уменьшать параллелизм динамически, а не увеличивать. Таким образом, на шагах 1 и 2 Pig переоценивает параллелизм.

Следующий параметр контролирует поведение автоматического распараллеливания в Tez (совпадает с MapReduce):

pig.exec.reducers.bytes.per.reducer
pig.exec.reducers.max

Изменение API

Если вызываете Pig в Java, происходит изменение PigStats и PigProgressNotificationListener при использовании PigRunner.run(), см. Статистику Pig и Слушатель уведомлений о ходе выполнения Pig

Известные проблемы

К текущим известным проблемам в режиме Tez относятся:

• Режим Tez в локальном режиме не стабилен; в некоторых случаях мы видим зависание задач.
• Специальный интерфейс пользователя для Tez пока недоступен; нет интерфейса для отслеживания хода выполнения задач. Однако сообщения в журнале доступны в интерфейсе.

Первым шагом повышения производительности и эффективности является измерение затрат времени. Pig предоставляет лёгкий метод приблизительного измерения времени, затрачиваемого в различных пользовательских функциях (UDFs) и загрузчиках. Просто установите свойство pig.udf.profile в значение true. Это позволит отслеживать новые счётчики для всех задач Map-Reduce, сгенерированных вашим скриптом: approx_microsecs измеряет приблизительное время, потраченное в UDF, а approx_invocations — приблизительное количество вызовов UDF. Кроме того, частота профилирования может быть настроена с помощью pig.udf.profile.frequency (по умолчанию — каждый 100-й вызов). Обратите внимание, что это может привести к большому количеству счётчиков (два на UDF). Избыточное количество счётчиков может привести к снижению производительности JobTracker, поэтому используйте эту функцию с осторожностью, желательно на тестовом кластере.

Комбинатор Pig — это оптимизатор, который вызывается, когда операторы в ваших скриптах расположены определённым образом. Примеры ниже демонстрируют, когда комбинатор используется, а когда нет. Всякий раз, когда это возможно, следует использовать комбинатор, так как он часто обеспечивает улучшение производительности на порядок.

Когда используется комбинатор

Комбинатор обычно используется в случае невложенного оператора foreach, где все проекции являются либо выражениями над столбцом группы, либо выражениями над алгебраическими UDF (см. Превратите свои UDF в алгебраические).

Пример:

A = load 'studenttab10k' as (name, age, gpa);
B = group A by age;
C = foreach B generate ABS(SUM(A.gpa)), COUNT(org.apache.pig.builtin.Distinct(A.name)), (MIN(A.gpa) + MAX(A.gpa))/2, group.age;
explain C;

В приведённом примере:

• Оператор GROUP можно рассматривать как единое целое или получать доступ к отдельным полям (как в примере).
• Оператор GROUP и его элементы могут находиться в любом месте в проекции.

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

• Функция преобразования столбца, например ABS, может применяться к алгебраической функции SUM.
• Алгебраическая функция (COUNT) может применяться к другой алгебраической функции (Distinct), но только внутренняя функция вычисляется с использованием комбинатора.
• Математическое выражение может применяться к одной или нескольким алгебраическим функциям.

Вы можете проверить, используется ли комбинатор для вашей запроса, выполнив EXPLAIN для псевдонима FOREACH, как показано выше. Вы должны увидеть раздел combine в части MapReduce плана:

.....
Combine Plan
B: Local Rearrange[tuple]{bytearray}(false) - scope-42
| |
| Project[bytearray][0] - scope-43
|
|---C: New For Each(false,false,false)[bag] - scope-28
| |
| Project[bytearray][0] - scope-29
| |
| POUserFunc(org.apache.pig.builtin.SUM$Intermediate)[tuple] - scope-30
| |
| |---Project[bag][1] - scope-31
| |
| POUserFunc(org.apache.pig.builtin.Distinct$Intermediate)[tuple] - scope-32
| |
| |---Project[bag][2] - scope-33
|
|---POCombinerPackage[tuple]{bytearray} - scope-36--------
.....

Комбинатор также используется с вложенным foreach, если единственной вложенной операцией является DISTINCT (см. FOREACH и Пример: Вложенный блок).

A = load 'studenttab10k' as (name, age, gpa);
B = group A by age;
C = foreach B { D = distinct (A.name); generate group, COUNT(D);}

Наконец, использование комбинатора зависит от окружения операторов GROUP и FOREACH.

Когда комбинатор не используется

Комбинатор обычно не используется, если между операторами GROUP и FOREACH в плане выполнения имеется какой-либо оператор. Даже если операторы находятся рядом в вашем скрипте, оптимизатор может их переупорядочить. В данном примере оптимизатор переместит FILTER выше FOREACH, что предотвратит использование комбинатора:

A = load 'studenttab10k' as (name, age, gpa);
B = group A by age;
C = foreach B generate group, COUNT (A);
D = filter C by group.age <30;

Обратите внимание, что скрипт можно сделать более эффективным, выполнив фильтрацию до оператора GROUP:

A = load 'studenttab10k' as (name, age, gpa);
B = filter A by age <30;
C = group B by age;
D = foreach C generate group, COUNT (B);

Примечание: Исключением из вышеуказанного правила является LIMIT. Начиная с Pig 0.9, даже если LIMIT находится между GROUP и FOREACH, комбинатор по-прежнему будет использоваться. В данном примере оптимизатор переместит LIMIT выше FOREACH, но это не предотвратит использование комбинатора.

A = load 'studenttab10k' as (name, age, gpa);
B = group A by age;
C = foreach B generate group, COUNT (A);
D = limit C 20;

Комбинатор также не используется в случае, когда несколько операторов FOREACH связаны с одним оператором GROUP:

A = load 'studenttab10k' as (name, age, gpa);
B = group A by age;
C = foreach B generate group, COUNT (A);
D = foreach B generate group, MIN (A.gpa). MAX(A.gpa);
.....

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

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

Включение/Выключение

Показано, что агрегация на основе хэширования увеличивает скорость операций группировки до 50%. Однако, так как это очень новая функция, она по умолчанию выключена. Чтобы включить её, установите свойство pig.exec.mapPartAgg в значение true.

Настройка

Если ключи группировки, используемые для группировки, не приводят к достаточной редукции числа записей, производительность может быть хуже с этой функцией, включённой. Чтобы этого не произошло, функция отключается сама по себе, если снижение числа записей, отправляемых комбинатору, не превышает настраиваемый порог. Этот порог может быть установлен с помощью свойства pig.exec.mapPartAgg.minReduction. По умолчанию он установлен в 10, что означает, что число записей, отправляемых комбинатору, должно быть уменьшено в 10 или более раз.

Pig выделяет фиксированное количество памяти для хранения пакетов и производит сброс на диск, как только лимит памяти достигается. Это очень похоже на то, как Hadoop решает, когда производить сброс данных, накопленных комбинатором.

Количество памяти, выделенное для пакетов, определяется свойством pig.cachedbag.memusage; значение по умолчанию составляет 20% (0.2) доступной памяти. Обратите внимание, что эта память разделяется между всеми большими пакетами, используемыми приложением.

По умолчанию Pig определяет число редьюсеров для конкретной задачи, основываясь на размере входных данных для фазы Map. Размер входных данных делится на значение параметра pig.exec.reducers.bytes.per.reducer (значение по умолчанию 1 ГБ) для определения количества редьюсеров. Максимальное число редьюсеров для задачи ограничено параметром pig.exec.reducers.max (значение по умолчанию 999).

Алгоритм оценивания редьюсеров по умолчанию, описанный выше, может быть переопределён путём установки параметра pig.exec.reducer.estimator в полное имя класса реализации org.apache.pig.backend.hadoop.executionengine.mapReduceLayer.PigReducerEstimator(MapReduce) или org.apache.pig.backend.hadoop.executionengine.tez.plan.optimizer.TezOperDependencyParallelismEstimator(Tez). Класс должен существовать в классе пути процесса, отправляющего задачу Pig. Если установлен параметр pig.exec.reducer.estimator.arg, значение будет передано в конструктор класса-реализации, принимающий единственный String.

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

Включение или отключение

Многозадачное выполнение включено по умолчанию. Чтобы отключить его и вернуться к поведению Pig «выполнять при dump/store», используйте опции «-M» или «-no_multiquery».

Чтобы запустить скрипт «myscript.pig» без оптимизации, выполните Pig следующим образом:

$ pig -M myscript.pig
or
$ pig -no_multiquery myscript.pig

Как это работает

Многозадачное выполнение вносит некоторые изменения:

• Для выполнения в пакетном режиме весь скрипт сначала анализируется, чтобы определить, можно ли объединить промежуточные задачи для уменьшения общего объема работы; выполнение начинается только после завершения анализа (см. оператор EXPLAIN и команды run и exec).

• Оптимизированы два сценария выполнения, как описано ниже: явные и неявные разделения и сохранение промежуточных результатов.

Явные и неявные разделения

Возможны случаи, когда вам нужно разное обработка разных частей одного потока данных.

Пример 1:

A = LOAD ...
...
SPLIT A' INTO B IF ..., C IF ...
...
STORE B' ...
STORE C' ...

Пример 2:

A = LOAD ...
...
B = FILTER A' ...
C = FILTER A' ...
...
STORE B' ...
STORE C' ...

В предыдущих версиях Pig Пример 1 выведет A' на диск, а затем запустит задачи для B' и C'. Пример 2 выполнит все зависимости B' и сохранит их, затем выполнит все зависимости C' и сохранит их. Оба варианта эквивалентны, но производительность будет разной.

Вот что делает многозадачное выполнение для повышения производительности:

• В Примере 2 добавляется неявное разделение для преобразования запроса в Пример 1. Это устраняет многократную обработку A'.

• Делает разделение неблокирующим и позволяет продолжить обработку. Это помогает уменьшить количество данных, которые необходимо хранить непосредственно при разделении.

• Разрешает несколько выходов из задачи. Таким образом, некоторые результаты могут быть сохранены как побочный эффект основной задачи. Это также необходимо для работы предыдущего пункта.

• Разрешает передавать несколько ветвей разделения в объединитель/редуктор. Это снова уменьшает объем ввода-вывода в случае, когда несколько ветвей в разделении могут извлечь выгоду из выполнения объединителя.

Сохранение промежуточных результатов

Иногда необходимо сохранять промежуточные результаты.

A = LOAD ...
...
STORE A'
...
STORE A''

Если скрипт не загружает A' повторно для обработки A, шаги выше A' будут дублироваться. Это особый случай Примера 2 выше, поэтому рекомендуется те же шаги. При многозадачном выполнении скрипт обработает A и сохранит A' как побочный результат.

Store vs. Dump

При многозадачном выполнении вы хотите использовать STORE для сохранения (сохранения) результатов. Не следует использовать DUMP, так как он отключит многозадачное выполнение и, вероятно, замедлит выполнение. (Если вы включили команды DUMP в свои скрипты для отладки, удалите их.)

Пример DUMP: в этом скрипте из-за того, что команда DUMP интерактивная, многозадачное выполнение будет отключено, и будут созданы две отдельные задачи для выполнения этого скрипта. Первая задача выполнит A > B > DUMP, а вторая задача выполнит A > B > C > STORE.

A = LOAD 'input' AS (x, y, z);
B = FILTER A BY x > 5;
DUMP B;
C = FOREACH B GENERATE y, z;
STORE C INTO 'output';

Пример STORE: в этом скрипте многозадачная оптимизация запустится, что позволит выполнить весь скрипт как одну задачу. Получаются два выходных значения: output1 и output2.

A = LOAD 'input' AS (x, y, z);
B = FILTER A BY x > 5;
STORE B INTO 'output1';
C = FOREACH B GENERATE y, z;
STORE C INTO 'output2';	

Обработка ошибок

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

Во-первых, Pig регистрирует все успешные и неудачные команды store. Команды store идентифицируются по пути вывода. В конце выполнения строка сводки указывает успех, частичный сбой или сбой всех команд store.

Во-вторых, Pig возвращает разные коды по завершении для этих сценариев:

• Код возврата 0: Все задачи завершились успешно

• Код возврата 1: Используется для исправляемых ошибок

• Код возврата 2: Все задачи завершились неудачно

• Код возврата 3: Некоторые задачи завершились неудачно

В некоторых случаях может быть целесообразно завершить весь скрипт при обнаружении первой завершившейся с ошибкой задачи. Этого можно добиться с помощью командной строки «-F» или «-stop_on_failure». При использовании Pig остановит выполнение при обнаружении первой завершившейся с ошибкой задачи и прекратит дальнейшую обработку. Это также означает, что команды файлов, которые следуют за неудачным store в скрипте, не будут выполняться (это можно использовать для создания файлов «done»).

Вот как используется флаг:

$ pig -F myscript.pig
or
$ pig -stop_on_failure myscript.pig

Обратная совместимость

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

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

В режиме map-reduce следующий скрипт загрузит данные из «hdfs://<host>:<port>/data1» и сохранит их в «hdfs://<host>:<port>/tmp/out1».

cd /;
A = LOAD 'data1';
cd tmp;
STORE A INTO 'out1';

Эти расширенные пути будут переданы любому реализатору LoadFunc или Slicer. В некоторых случаях это может вызвать проблемы, особенно когда LoadFunc/Slicer не используется для чтения из файла dfs или пути (например, загрузки из базы данных SQL).

Решения:

• Укажите «-M» или «-no_multiquery», чтобы вернуться к старым именам

• Укажите пользовательскую схему для LoadFunc/Slicer

Аргументы, используемые в операторе LOAD, имеющие схему, отличную от «hdfs» или «file», не будут расширены и будут переданы LoadFunc/Slicer без изменений.

В случае SQL функция SQLLoader вызывается с 'sql://mytable'.

A = LOAD 'sql://mytable' USING SQLLoader();

Неявные зависимости

Если скрипт имеет зависимости от порядка выполнения за пределами того, что Pig знает, выполнение может завершиться неудачно.

Пример

В этом скрипте MYUDF может попытаться прочитать из out1, файла, в который A был только что сохранён. Однако Pig не знает, что MYUDF зависит от файла out1 и может отправить задачи, создающие файлы out2 и out1 одновременно.

...
STORE A INTO 'out1';
B = LOAD 'data2';
C = FOREACH B GENERATE MYUDF($0,'out1');
STORE C INTO 'out2';

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

...
STORE A INTO 'out1';
EXEC;
B = LOAD 'data2';
C = FOREACH B GENERATE MYUDF($0,'out1');
STORE C INTO 'out2';

Пример

В этом скрипте операторы STORE/LOAD имеют разные пути к файлам; однако оператор LOAD зависит от оператора STORE.

A = LOAD '/user/xxx/firstinput' USING PigStorage();
B = group ....
C = .... agrregation function
STORE C INTO '/user/vxj/firstinputtempresult/days1';
..
Atab = LOAD '/user/xxx/secondinput' USING  PigStorage();
Btab = group ....
Ctab = .... agrregation function
STORE Ctab INTO '/user/vxj/secondinputtempresult/days1';
..
E = LOAD '/user/vxj/firstinputtempresult/' USING  PigStorage();
F = group ....
G = .... aggregation function
STORE G INTO '/user/vxj/finalresult1';

Etab =LOAD '/user/vxj/secondinputtempresult/' USING  PigStorage();
Ftab = group ....
Gtab = .... aggregation function
STORE Gtab INTO '/user/vxj/finalresult2';

Чтобы заставить скрипт работать, добавьте оператор exec.

A = LOAD '/user/xxx/firstinput' USING PigStorage();
B = group ....
C = .... agrregation function
STORE C INTO '/user/vxj/firstinputtempresult/days1';
..
Atab = LOAD '/user/xxx/secondinput' USING  PigStorage();
Btab = group ....
Ctab = .... agrregation function
STORE Ctab INTO '/user/vxj/secondinputtempresult/days1';

EXEC;

E = LOAD '/user/vxj/firstinputtempresult/' USING  PigStorage();
F = group ....
G = .... aggregation function
STORE G INTO '/user/vxj/finalresult1';
..
Etab =LOAD '/user/vxj/secondinputtempresult/' USING  PigStorage();
Ftab = group ....
Gtab = .... aggregation function
STORE Gtab INTO '/user/vxj/finalresult2';

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

• Свойство pig.optimizer.rules.disabled свойства pig, которое принимает список оптимизационных правил, разделяемых запятыми, для отключения; ключевое слово all отключает все необязательные оптимизации. (Например: set pig.optimizer.rules.disabled 'ColumnMapKeyPrune';)
• Параметры командной строки -t, -optimizer_off. (Например: pig -optimizer_off [opt_rule | all])

FilterLogicExpressionSimplifier является исключением из вышесказанного. Правило отключено по умолчанию и включено путем установки свойства pig.exec.filterLogicExpressionSimplifier pig на true.

PartitionFilterOptimizer

Переместить условие фильтрации в загрузчик.

A = LOAD 'input' as (dt, state, event) using HCatLoader();
B = FILTER A BY dt=='201310' AND state=='CA';

Условие фильтрации будет перенесено в загрузчик, если загрузчик его поддерживает (обычно загрузчик ориентирован на разделы, такой как HCatLoader)

A = LOAD 'input' as (dt, state, event) using HCatLoader();
--Filter is removed

Загрузчик получит инструкцию загрузить раздел с dt=='201310' и state=='CA'

PredicatePushdownOptimizer

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

A = LOAD 'input' using OrcStorage();
B = FILTER A BY dt=='201310' AND state=='CA';

Условие фильтрации будет перенесено в загрузчик, если загрузчик его поддерживает

A = LOAD 'input' using OrcStorage();  -- Filter condition push to loader
B = FILTER A BY dt=='201310' AND state=='CA';  -- Filter evaluated in Pig again

ConstantCalculator

Это правило вычисляет константное выражение.

1) Constant pre-calculation 

B = FILTER A BY a0 > 5+7; 
is simplified to 
B = FILTER A BY a0 > 12; 

2) Evaluate UDF

B = FOREACH A generate UPPER(CONCAT('a', 'b'));
is simplified to 
B = FOREACH A generate 'AB';

SplitFilter

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

A = LOAD 'input1' as (a0, a1);
B = LOAD 'input2' as (b0, b1);
C = JOIN A by a0, B by b0;
D = FILTER C BY a1>0 and b1>0;

Здесь D будет разделен на:

X = FILTER C BY a1>0;
D = FILTER X BY b1>0;

Таким образом, "a1>0" и "b1>0" могут быть перенесены индивидуально.

PushUpFilter

Цель этого правила — перенести операторы FILTER вверх по графу потока данных. В результате уменьшается количество записей, проходящих через конвейер.

A = LOAD 'input';
B = GROUP A BY $0;
C = FILTER B BY $0 < 10;

MergeFilter

Объединить условия фильтрации после правила PushUpFilter для уменьшения числа операторов фильтрации.

PushDownForEachFlatten

Цель этого правила — уменьшить количество записей, проходящих через конвейер, путем перемещения операторов FOREACH с FLATTEN вниз по графу потока данных. В примере ниже было бы эффективнее перенести foreach после объединения, чтобы уменьшить стоимость операции объединения.

A = LOAD 'input' AS (a, b, c);
B = LOAD 'input2' AS (x, y, z);
C = FOREACH A GENERATE FLATTEN($0), B, C;
D = JOIN C BY $1, B BY $1;

LimitOptimizer

Цель этого правила — перенести оператор LIMIT вверх по графу потока данных (или вниз по дереву для баз данных). Кроме того, для top-k (ORDER BY, за которым следует LIMIT) LIMIT переносится в ORDER BY.

A = LOAD 'input';
B = ORDER A BY $0;
C = LIMIT B 10;

ColumnMapKeyPrune

Упростить загрузчик, чтобы загружать только необходимые столбцы. Прирост производительности больше, если соответствующий загрузчик поддерживает обрезку столбцов и загружает только необходимые столбцы (см. LoadPushDown.pushProjection). В противном случае ColumnMapKeyPrune вставит оператор ForEach сразу после загрузчика.

A = load 'input' as (a0, a1, a2);
B = ORDER A by a0;
C = FOREACH B GENERATE a0, a1;

a2 не относится к этому запросу, поэтому мы можем обрезать его раньше. Загрузчик в этом запросе — PigStorage, и он поддерживает обрезку столбцов. Поэтому мы загружаем только a0 и a1 из входного файла.

ColumnMapKeyPrune также обрезает неиспользуемые ключи карты:

A = load 'input' as (a0:map[]);
B = FOREACH A generate a0#'key1';

AddForEach

Обрезать неиспользуемый столбец как можно скорее. В дополнение к обрезке загрузчика в ColumnMapKeyPrune, мы можем обрезать столбец, как только он не используется в остальной части сценария

-- Original code: 

A = LOAD 'input' AS (a0, a1, a2); 
B = ORDER A BY a0;
C = FILTER B BY a1>0;

Мы можем обрезать только a2 из загрузчика. Однако a0 никогда не используется после "ORDER BY". Поэтому мы можем удалить a0 сразу после оператора "ORDER BY".

-- Optimized code: 

A = LOAD 'input' AS (a0, a1, a2); 
B = ORDER A BY a0;
B1 = FOREACH B GENERATE a1;  -- drop a0
C = FILTER B1 BY a1>0;

MergeForEach

Цель этого правила — объединить два оператора foreach, если выполняются следующие условия:

• Операторы foreach являются последовательными.
• Первый оператор foreach не содержит flatten.
• Второй оператор foreach не является вложенным.

-- Original code: 

A = LOAD 'file.txt' AS (a, b, c); 
B = FOREACH A GENERATE a+b AS u, c-b AS v; 
C = FOREACH B GENERATE $0+5, v; 

-- Optimized code: 

A = LOAD 'file.txt' AS (a, b, c); 
C = FOREACH A GENERATE a+b+5, c-b;

GroupByConstParallelSetter

Принудительно установить параллелизм "1" для оператора "group all". Это связано с тем, что даже если мы установим параллелизм в N, в этом случае будет использоваться только 1 редуктор, а все остальные редукторы производят пустой результат.

A = LOAD 'input';
B = GROUP A all PARALLEL 10;

Реплицированные соединения

Соединение фрагмент-реплики — это специальный тип соединения, который хорошо работает, если одна или несколько отношений достаточно малы, чтобы поместиться в оперативную память. В таких случаях Pig может выполнить очень эффективное соединение, поскольку вся работа Hadoop выполняется на стороне Map. В этом типе соединения большое отношение следует за одним или несколькими небольшими отношениями. Небольшие отношения должны быть достаточно малыми, чтобы поместиться в оперативную память; если это не так, процесс завершается ошибкой.

Использование

Выполните реплицированное соединение с помощью предложения USING (см. JOIN (внутреннее) и JOIN (внешнее)). В этом примере большое отношение соединяется с двумя меньшими отношениями. Обратите внимание, что большое отношение идёт первым, а за ним следует меньшие; и все меньшие отношения вместе должны поместиться в оперативную память, в противном случае генерируется ошибка.

big = LOAD 'big_data' AS (b1,b2,b3);

tiny = LOAD 'tiny_data' AS (t1,t2,t3);

mini = LOAD 'mini_data' AS (m1,m2,m3);

C = JOIN big BY b1, tiny BY t1, mini BY m1 USING 'replicated';

Условия

Соединения фрагмент-реплики — экспериментальные; у нас нет чёткого представления о том, насколько маленьким должно быть малое отношение, чтобы поместиться в память. В наших тестах с простым запросом, включающим только JOIN, можно использовать отношение объёмом до 100 МБ, если весь процесс получает 1 ГБ памяти. Пожалуйста, поделитесь своими наблюдениями и опытом с нами.

Чтобы избежать реплицированных соединений с большими отношениями, мы завершаем работу с ошибкой, если размер отношения (отношений), подлежащего репликации (в байтах), больше, чем pig.join.replicated.max.bytes (по умолчанию = 1 ГБ).

Несбалансированные соединения

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

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

Использование

Выполните несбалансированное соединение с помощью предложения USING (см. JOIN (внутреннее) и JOIN (внешнее)).

A = LOAD 'skewed_data' AS (a1,a2,a3);
B = LOAD 'data' AS (b1,b2,b3);
C = JOIN A BY a1, B BY b1 USING 'skewed';

Условия

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

• Несбалансированное соединение работает с внутренними и внешними соединениями двух таблиц. В настоящее время мы не поддерживаем более двух таблиц для несбалансированного соединения. Указание соединений с тремя (или более) таблицами приведёт к отказу проверки. Для таких соединений мы полагаемся на вас, чтобы вы разбили их на соединения двух таблиц.
• Таблица, подверженная несбалансированности, должна быть указана в качестве левой таблицы. Pig берёт образец этой таблицы и определяет количество редьюсеров на ключ.
• Параметр Java pig.skewedjoin.reduce.memusage задаёт долю доступной кучи для редьюсера для выполнения соединения. Низкая доля заставляет Pig использовать больше редьюсеров, но увеличивает стоимость копирования. Мы наблюдали хорошую производительность, когда устанавливали это значение в диапазоне 0,1-0,4. Однако обратите внимание, что это вряд ли точный диапазон. Его значение зависит от объёма кучи, доступной для операции, количества столбцов в входе и степени несбалансированности. Наиболее подходящее значение лучше всего получить, проведя эксперименты для достижения хорошей производительности. Значение по умолчанию — 0,5.
• Несбалансированное соединение не решает (не уравновешивает) неравномерное распределение данных по редьюсерам. Однако в большинстве случаев несбалансированное соединение гарантирует, что соединение завершится (хотя и медленно), а не завершится ошибкой.

Соединения слиянием

Часто данные пользователя хранятся таким образом, что оба входа уже отсортированы по ключу соединения. В этом случае можно соединить данные на стадии Map в задаче MapReduce. Это обеспечивает значительное повышение производительности по сравнению с передачей всех данных через ненужные стадии сортировки и перемешивания.

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

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

Каждый map использует индекс для поиска соответствующей записи в правом входе и начинает выполнение соединения.

Использование

Выполните соединение слиянием с помощью предложения USING (см. JOIN (внутреннее) и JOIN (внешнее)).

C = JOIN A BY a1, B BY b1, C BY c1 USING 'merge';

Условия

Условие A

Внутреннее соединение слиянием (между двумя таблицами) будет работать только при следующих условиях:

• Данные должны поступать непосредственно из оператора Load или Order.
• Между источником отсортированных данных и оператором соединения могут быть операторы фильтрации и foreach. Оператор foreach должен соответствовать следующим условиям:
  • Оператор foreach не должен изменять положение ключей соединения.
  • Не должно быть преобразований над ключами соединения, которые изменят порядок сортировки.
  • UDFs также должны соответствовать предыдущему условию и не должны преобразовывать ключи JOIN таким образом, чтобы изменить порядок сортировки.
• Данные должны быть отсортированы по ключам соединения в порядке возрастания (ASC) с обеих сторон.
• Если сортировка предоставлена загрузочником, а не явным оператором Order, загрузочник правой стороны должен реализовывать интерфейс {OrderedLoadFunc} или интерфейс {IndexableLoadFunc}.
• Информация о типе должна быть предоставлена для ключа соединения в схеме.

Загрузочник PigStorage удовлетворяет всем этим условиям.

Условие B

Внешнее соединение слиянием (между двумя таблицами) и внутреннее соединение слиянием (между тремя или более таблицами) будут работать только при следующих условиях:

• Другие операции не могут выполняться между операторами загрузки и соединения.
• Данные должны быть отсортированы по ключам соединения в порядке возрастания (ASC) с обеих сторон.
• Загрузочник левой стороны должен реализовывать интерфейс {CollectableLoader}, а также {OrderedLoadFunc}.
• Все остальные загрузочники должны реализовывать интерфейс {IndexableLoadFunc}.
• Информация о типе должна быть предоставлена для ключа соединения в схеме.

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

Соединения слиянием-разреженными данными

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

Использование

Выполните соединение слиянием-разреженными данными с помощью предложения USING (см. JOIN (внутреннее)).

a = load 'sorted_input1' using org.apache.pig.piggybank.storage.IndexedStorage('\t', '0');
b = load 'sorted_input2' using org.apache.pig.piggybank.storage.IndexedStorage('\t', '0');
c = join a by $0, b by $0 using 'merge-sparse';
store c into 'results';

Условия

Соединение слиянием-разреженными данными работает только для внутренних соединений и в настоящее время не реализовано для внешних соединений.

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

Piggybank теперь содержит функцию загрузки org.apache.pig.piggybank.storage.IndexedStorage, которая является производной от PigStorage и реализует IndexedLoadFunc. Это единственный загрузочник, включенный в стандартное распределение Pig, который можно использовать для соединения слиянием-разреженными данными.

Соображения по производительности

Обратите внимание на следующее:

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