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

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

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

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;

.....
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--------
.....

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

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;

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);

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

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);
.....

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

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

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

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

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';

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 -F myscript.pig
or
$ pig -stop_on_failure myscript.pig

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

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

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

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

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';

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';

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

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

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

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';

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

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

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

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;

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

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

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

-- Original code: 

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

-- 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;

-- 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;

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

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

Pig поддерживает различные правила оптимизации, которые включены по умолчанию. Ознакомьтесь с этими правилами.

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

Если типы не указаны в операторе загрузки, Pig предполагает тип =double= для числовых вычислений. Часто ваши данные будут значительно меньше, возможно, целые числа или длинные целые числа. Указание реального типа поможет ускорить арифметические вычисления. Это также предоставляет дополнительное преимущество – раннее обнаружение ошибок.

--Query 1
A = load 'myfile' as (t, u, v);
B = foreach A generate t + u;

--Query 2
A = load 'myfile' as (t: int, u: int, v);
B = foreach A generate t + u;

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

Ранняя и частая проекция

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

A = load 'myfile' as (t, u, v);
B = load 'myotherfile' as (x, y, z);
C = join A by t, B by x;
D = group C by u;
E = foreach D generate group, COUNT($1);

В этом запросе нет необходимости использовать v, y или z. И нет необходимости передавать как t, так и x после объединения, достаточно одного из них. Изменение приведенного выше запроса на запрос ниже значительно уменьшит количество данных, передаваемых через фазы map и reduce в Pig.

A = load 'myfile' as (t, u, v);
A1 = foreach A generate t, u;
B = load 'myotherfile' as (x, y, z);
B1 = foreach B generate x;
C = join A1 by t, B1 by x;
C1 = foreach C generate t, u;
D = group C1 by u;
E = foreach D generate group, COUNT($1);

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

Раннее и частое применение фильтров

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

-- Query 1
A = load 'myfile' as (t, u, v);
B = load 'myotherfile' as (x, y, z);
C = filter A by t == 1;
D = join C by t, B by x;
E = group D by u;
F = foreach E generate group, COUNT($1);

-- Query 2
A = load 'myfile' as (t, u, v);
B = load 'myotherfile' as (x, y, z);
C = join A by t, B by x;
D = group C by u;
E = foreach D generate group, COUNT($1);
F = filter E by C.t == 1;

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

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

Сокращение конвейера операторов

Для большей ясности вашего скрипта вы можете разделить свои проекты на несколько шагов, например:

A = load 'data' as (in: map[]);
-- get key out of the map
B = foreach A generate in#'k1' as k1, in#'k2' as k2;
-- concatenate the keys
C = foreach B generate CONCAT(k1, k2);
.......

Хотя приведенный выше пример легче читать, вы можете рассмотреть возможность объединения двух операторов foreach для повышения производительности запроса:

A = load 'data' as (in: map[]);
-- concatenate the keys from the map
B = foreach A generate CONCAT(in#'k1', in#'k2');
....

То же самое относится к фильтрам.

Делайте ваши UDF алгебраическими

Запросы, которые могут использовать комбинирующий этап, как правило, выполняются значительно быстрее (иногда в несколько раз быстрее), чем версии, которые этого не делают. Последний код существенно улучшает использование комбинирующего этапа; однако, вам необходимо внести свой вклад. Если у вас есть UDF, работающий с группированными данными и по своей природе алгебраический (то есть его вычисления можно разбить на несколько шагов), убедитесь, что вы реализовали его как таковой. Подробности о написании алгебраических UDF см. в разделе Алгебраический интерфейс.

A = load 'data' as (x, y, z)
B = group A by x;
C = foreach B generate group, MyUDF(A);
....

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

Использование интерфейса аккумулятора

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

Примечание: Pig автоматически выбирает интерфейс, который, по его мнению, обеспечит наилучшую производительность: Алгебраический > Аккумулятор > По умолчанию.

Удаление нулевых значений перед объединением

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

Это объединение

A = load 'myfile' as (t, u, v);
B = load 'myotherfile' as (x, y, z);
C = join A by t, B by x;

переписывается Pig в

A = load 'myfile' as (t, u, v);
B = load 'myotherfile' as (x, y, z);
C1 = cogroup A by t INNER, B by x INNER;
C = foreach C1 generate flatten(A), flatten(B);

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

Если запрос переписан в

A = load 'myfile' as (t, u, v);
B = load 'myotherfile' as (x, y, z);
A1 = filter A by t is not null;
B1 = filter B by x is not null;
C = join A1 by t, B1 by x;

тогда нулевые значения будут удалены до объединения. Поскольку все нулевые ключи отправляются одному редуктору, если ваш ключ равен нулю даже небольшой процент времени, выгода может быть значительной. В одном тесте, где ключ был нулевым в 7% случаев, а данные были распределены по 200 редукторам, мы наблюдали примерно 10-кратное ускорение запроса путем добавления ранних фильтров.

Использование оптимизаций для объединения

Оптимизации обычного объединения

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

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

small = load 'small_file' as (t, u, v);
large = load 'large_file' as (x, y, z);
C = join small by t, large by x;

Специализированные оптимизации объединения

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

Использование параллельных возможностей

Вы можете установить количество задач reduce для MapReduce-задач, генерируемых Pig, используя две параллельные возможности. (Параллельные возможности влияют только на количество задач reduce. Параллелизм map определяется входным файлом, одна карта для каждого блока HDFS.)

Вы устанавливаете количество редукторов

Используйте команду set default parallel, чтобы установить количество редукторов на уровне скрипта.

В качестве альтернативы используйте оператор PARALLEL для установки количества редукторов на уровне оператора. (В скрипте значение, установленное с помощью оператора PARALLEL, переопределяет любое значение, установленное с помощью «set default parallel»). Вы можете включить оператор PARALLEL с любым оператором, начинающим фазу reduce: COGROUP, CROSS, DISTINCT, GROUP, JOIN (внутреннее), JOIN (внешнее) и ORDER BY. Оператор PARALLEL также может использоваться с UNION, если режим выполнения – Tez. Он отключит оптимизацию union и добавит дополнительный шаг reduce. Хотя производительность будет немного ниже из-за дополнительного шага, это очень полезно для управления количеством выходных файлов.

Количество редукторов, необходимое для конкретной конструкции в Pig, образующей границу MapReduce, полностью зависит от (1) ваших данных и количества промежуточных ключей, которые генерируются вашими mappперами, и (2) от партиционера и распределения ключей выходных данных map (combiner). В лучших случаях мы наблюдали, что редуктор, обрабатывающий около 1 ГБ данных, работает эффективно.

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

Если ни «set default parallel», ни оператор PARALLEL не используются, Pig устанавливает количество редукторов с использованием эвристики, основанной на размере входных данных. Вы можете установить значения для этих свойств:

• pig.exec.reducers.bytes.per.reducer - Определяет количество входных байтов на редуктор; значение по умолчанию составляет 1000*1000*1000 (1 ГБ).
• pig.exec.reducers.max - Определяет верхнюю границу количества редукторов; значение по умолчанию – 999.

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

#reducers = MIN (pig.exec.reducers.max, общее количество входных данных (в байтах) / байты на редуктор)

Примеры

В этом примере оператор PARALLEL используется с оператором GROUP.

A = LOAD 'myfile' AS (t, u, v);
B = GROUP A BY t PARALLEL 18;
...

В этом примере все запускаемые MapReduce-задачи используют 20 редукторов.

SET default_parallel 20;
A = LOAD 'myfile.txt' USING PigStorage() AS (t, u, v);
B = GROUP A BY t;
C = FOREACH B GENERATE group, COUNT(A.t) as mycount;
D = ORDER C BY mycount;
STORE D INTO 'mysortedcount' USING PigStorage();

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

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

Выборка:

A = load 'myfile' as (t, u, v);
B = limit A 500;

Лучшие результаты:

A = load 'myfile' as (t, u, v);
B = order A by t;
C = limit B 500;

Предпочтите DISTINCT оператору GROUP BY/GENERATE

Для извлечения уникальных значений из столбца в отношении можно использовать DISTINCT или GROUP BY/GENERATE. DISTINCT предпочтительнее; он быстрее и эффективнее.

Пример с использованием GROUP BY - GENERATE:

A = load 'myfile' as (t, u, v);
B = foreach A generate u;
C = group B by u;
D = foreach C generate group as uniquekey;
dump D; 

Пример с использованием DISTINCT:

A = load 'myfile' as (t, u, v);
B = foreach A generate u;
C = distinct B;
dump C; 

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

Если ваш скрипт Pig генерирует последовательность MapReduce-задач, вы можете сжать вывод промежуточных заданий с помощью сжатия LZO. (Используйте оператор EXPLAIN, чтобы определить, создает ли ваш скрипт несколько MapReduce-задач.)

Это позволит сэкономить пространство HDFS, используемое для хранения промежуточных данных, используемых PIG, и потенциально повысить скорость выполнения запроса. Чем больше промежуточных данных генерируется, тем больше выгоды в плане хранения и скорости.

Вы можете установить значения для этих свойств:

• pig.tmpfilecompression - Определяет, следует ли сжимать временные файлы или нет (по умолчанию установлено в false).
• pig.tmpfilecompression.codec - Указывает используемый кодек сжатия. В настоящее время Pig принимает «gz» и «lzo» в качестве возможных значений. Однако, поскольку LZO распространяется под лицензией GPL (и отключен по умолчанию), вам необходимо настроить свой кластер для использования кодека LZO, чтобы воспользоваться этой функцией. Для получения подробностей см. http://code.google.com/p/hadoop-gpl-compression/wiki/FAQ.

В нетривиальных запросах (один из них выполнялся более пары минут) мы наблюдали значительные улучшения как в задержке запроса, так и в использовании пространства. Для некоторых запросов мы наблюдали до 96% экономии дискового пространства и до 4-кратного ускорения запроса. Конечно, характеристики производительности сильно зависят от запроса и данных, и необходимо проводить тестирование, чтобы определить полученные преимущества. Мы не наблюдали замедления в проведенных тестах, что означает, что вы, по крайней мере, экономите место, используя сжатие.

С gzip мы наблюдали лучшее сжатие (96-99%), но ценой замедления на 4%. Поэтому мы не рекомендуем использовать gzip.

Пример

-- launch Pig script using lzo compression 

java -cp $PIG_HOME/pig.jar 
-Djava.library.path=<path to the lzo library> 
-Dpig.tmpfilecompression=true 
-Dpig.tmpfilecompression.codec=lzo org.apache.pig.Main  myscript.pig 

Объединение небольших входных файлов

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

Вы можете установить значения для этих свойств:

• pig.maxCombinedSplitSize – Указывает размер данных в байтах, обрабатываемых одной картой. Более мелкие файлы объединяются до достижения этого размера.
• pig.splitCombination – Включает или отключает объединение файлов разделов (по умолчанию установлено «true»).

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

• Если ваша реализация загрузчика использует объект PigSplit, переданный через метод prepareToRead, вам может потребоваться перестроить загрузчик, так как определение PigSplit было изменено.
• Загрузчик должен быть бессостоятельным при вызовах метода prepareToRead. То есть, метод должен сбросить любые внутренние состояния, которые не зависят от аргумента RecordReader.
• Если загрузчик реализует IndexableLoadFunc или реализует OrderedLoadFunc и CollectableLoadFunc, его входные разделы не будут подвержены возможным объединениям.

Прямой запрос

Когда оператор DUMP используется для выполнения операторов Pig Latin, Pig может минимизировать задержку, напрямую считывая данные из HDFS, вместо запуска MapReduce задач.

Результат извлекается, если запрос содержит любой из следующих операторов: FILTER, FOREACH, LIMIT, STREAM, UNION.
Извлечение будет отключено в случае:

• наличия других операторов, загрузчиков выборок и скалярных выражений
• отсутствия оператора LIMIT
• явных разделов

Также обратите внимание, что прямой запрос не поддерживает UDF, которые взаимодействуют с распределённым кешем. Вы можете проверить, можно ли выполнить запрос с помощью EXPLAIN. Вы должны увидеть «No MR jobs. Fetch only.» в части MapReduce плана.

Прямой запрос включён по умолчанию. Чтобы его отключить, установите свойство opt.fetch в false или запустите Pig с опцией "-N" или "-no_fetch".

Автоматический локальный режим

Обработка небольших задач MapReduce в кластере Hadoop может быть медленной из-за накладных расходов на запуск и планирование задач. Для задач с небольшими входными данными Pig может преобразовать их в выполнение в процессе MapReduce с локальным режимом Hadoop. Если флаг pig.auto.local.enabled установлен в true, Pig преобразует задачи MapReduce с входными данными меньше, чем pig.auto.local.input.maxbytes (по умолчанию 100 МБ), для выполнения в локальном режиме, при условии, что количество необходимых редукторов не превышает 1. Обратите внимание, что задачи, преобразованные для выполнения в локальном режиме, загружают и сохраняют данные из HDFS, поэтому любая задача в рабочем процессе Pig (DAG) может быть преобразована в локальный режим, не влияя на её последующие задачи.

Вы можете установить значения этих свойств для настройки поведения:

• pig.auto.local.enabled - Включает/выключает функцию автоматического локального режима (по умолчанию false).
• pig.auto.local.input.maxbytes - Управляет максимальным пороговым размером (в байтах) для преобразования задач в локальный режим (по умолчанию 100 МБ).

Иногда вам может потребоваться изменить конфигурацию задач, которые преобразуются в локальный режим (например, изменить io.sort.mb для небольших задач). Для этого можно использовать префикс pig.local. для любой конфигурации, и конфигурация будет установлена для преобразованных задач. Например, установка pig.local.io.sort.mb 100 изменит значение io.sort.mb на 100 для задач, преобразованных для выполнения в локальном режиме.

Кеш пользовательских JAR-файлов

JAR-файлы, необходимые для пользовательских функций (UDF), копируются в распределённый кэш Pig, чтобы сделать их доступными на узлах задач. Для размещения этих JAR-файлов в распределённом кэше клиенты Pig копируют их в HDFS по временному расположению. Для запланированных задач эти JAR-файлы не меняются часто. Кроме того, создание большого количества небольших JAR-файлов в HDFS не является удобным для HDFS. Чтобы избежать повторной копии этих маленьких JAR-файлов в HDFS, Pig позволяет пользователям настроить кэш JAR-файлов на уровне пользователя (доступен только пользователю по соображениям безопасности). Если флаг pig.user.cache.enabled установлен в true, JAR-файлы UDF копируются в расположение кэша JAR-файлов (настраиваемое) в каталоге, имя которого содержит хэш (SHA) JAR-файла. Хэш JAR-файла используется для идентификации существования JAR-файла при последующем использовании пользователем. Если JAR-файл с таким же хэшем и именем файла найден в кэше, он используется, избегая копирования JAR-файла в HDFS.

Вы можете установить значения этих свойств для настройки кэша JAR-файлов:

• pig.user.cache.enabled - Включает/выключает функцию кэширования JAR-файлов пользователя (по умолчанию false).
• pig.user.cache.location - Путь в HDFS, который будет использоваться как каталог для кэширования JAR-файлов пользователя (по умолчанию pig.temp.dir или /tmp).

Функция кэширования JAR-файлов пользователя является безопасной. Если JAR-файлы не могут быть скопированы в кэш JAR-файлов из-за проблем с правами доступа или конфигурацией, Pig вернётся к старому поведению.

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

Соединение фрагмента реплики — это особый тип соединения, который хорошо работает, если одна или несколько отношений достаточно малы, чтобы поместиться в оперативную память. В таких случаях 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 ГБ).

Соединения Bloom

Соединение Bloom — это особый тип соединения, в котором используется фильтр Bloom, созданный с помощью ключей соединения одного отношения, и используется для фильтрации записей других отношений перед выполнением обычного соединения с помощью хеширования. Объём данных, отправляемых в редьюсеры, будет значительно меньше в зависимости от количества записей, отфильтрованных на стороне map. Соединение Bloom очень полезно в тех случаях, когда количество совпадающих записей между отношениями в соединении сравнительно меньше по сравнению с общим числом записей, что позволяет отфильтровать многие из них перед соединением. До добавления соединения Bloom в качестве типа соединения пользователи достигали аналогичного функционала, используя встроенные UDF Bloom, что не так эффективно и требовало больше строк кода. В настоящее время соединение Bloom реализовано только в режиме выполнения Tez. Встроенные UDF Bloom необходимо использовать для других режимов выполнения.

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

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

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

medium = LOAD 'medium_data' AS (m1,m2,m3);

small = LOAD 'small_data' AS (s1,s2,s3);

C = JOIN big BY b1, medium BY m1, small BY s1 USING 'bloom';

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

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

small = LOAD 'small_data' AS (m1,m2,m3);

C = JOIN small BY s1 LEFT, big BY b1 USING 'bloom';

Условия

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

Параметры настройки

Существует множество свойств Pig, которые можно настроить для построения более эффективного фильтра Bloom. См. Bloom Filter для обсуждения выбора количества битов и количества хеш-функций. Более простой вариант — поискать в поисковой системе «калькулятор фильтра Bloom» и использовать один из доступных онлайн-калькуляторов, чтобы получить нужные значения.

• pig.bloomjoin.strategy — Допустимые значения для этого параметра — 'map' и 'reduce'. Значение по умолчанию — map. Соединение Bloom имеет два различных типа реализации, чтобы быть более эффективным в разных случаях. Как правило, в DAG есть дополнительный шаг reduce для построения фильтра(ов) Bloom.
  • map — В каждом map фильтры Bloom вычисляются по ключам соединения, разделяемые с помощью хешкода ключа с pig.bloomjoin.num.filters числом разделов. Фильтры Bloom для каждого раздела из разных map затем комбинируются в редьюсерах, создавая один фильтр Bloom на раздел. Значение по умолчанию pig.bloomjoin.num.filters для этой стратегии равно 1, и, следовательно, обычно создаётся только один фильтр Bloom. Это эффективно и быстро, если число map меньше ( < 10) и число уникальных ключей не слишком велико. Это может быть быстрее с большим числом map и даже с большими размерами вектора Bloom, но объём данных, перемещаемых в редьюсер для агрегирования, становится огромным, делая его неэффективным.
  • reduce — Ключи соединения отправляются из map в редьюсер, разделённые с помощью хешкода ключа с pig.bloomjoin.num.filters числом разделов. В редьюсерах затем вычисляется один фильтр Bloom на раздел. Число редьюсеров устанавливается равным числу разделов, что позволяет вычислять каждый фильтр Bloom параллельно. Значение по умолчанию pig.bloomjoin.num.filters для этой стратегии равно 11. Это эффективно для больших наборов данных с большим количеством map или очень большими размерами вектора Bloom. В этом случае объём ключей, отправленных в редьюсер, меньше, чем объём отправленных фильтров Bloom для агрегирования, что делает его эффективным.
• pig.bloomjoin.num.filters — Количество фильтров Bloom, которые будут созданы. По умолчанию 1 для стратегии map и 11 для стратегии reduce.
• pig.bloomjoin.vectorsize.bytes — Размер в байтах битового вектора, используемого для фильтра Bloom. Больший размер вектора потребуется при большем количестве уникальных ключей. Значение по умолчанию — 1048576 (1 МБ).
• pig.bloomjoin.hash.functions — Тип хеш-функции для использования. Допустимые значения — 'jenkins' и 'murmur'. По умолчанию — murmur.
• pig.bloomjoin.hash.types — Количество хеш-функций, используемых при вычислении фильтра Bloom. Определяет вероятность ложных срабатываний. Чем больше значение, тем меньше ложных срабатываний. Слишком большое значение может увеличить время работы процессора. Значение по умолчанию — 3.

Соединения с неравномерным распределением

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

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

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

Выполните соединение с неравномерным распределением с помощью предложения 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 не должен менять положение ключей соединения.
  • Не должно быть преобразований ключей соединения, которые изменят порядок сортировки.
  • UDF также должны соответствовать предыдущему условию и не должны преобразовывать ключи 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, который может быть использован для соединения слиянием-разреженными данными.

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

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

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