Apache Pig: exécution parallèle et aplatie de réducteurs

j'ai implémenté un script de cochon Apache. Quand j'exécute le script, il en résulte de nombreux mappers pour une étape spécifique, mais n'a qu'un réducteur pour cette étape. En raison de cette condition (beaucoup de mappers, un réducteur) le cluster Hadoop est presque inactif tandis que le réducteur simple exécute. Afin de mieux utiliser les ressources du cluster, je voudrais également avoir de nombreux réducteurs fonctionnant en parallèle.

même si je mets le parallélisme dans le script Pig en utilisant le paramètre DEFAULT_PARALLEL commande j'ai toujours raison dans le fait d'avoir seulement 1 réducteur.

La partie de code de délivrer le problème est le suivant:

SET DEFAULT_PARALLEL 5;
inputData = LOAD 'input_data.txt' AS (group_name:chararray, item:int);
inputDataGrouped = GROUP inputData BY (group_name);
-- The GeneratePairsUDF generates a bag containing pairs of integers, e.g. {(1, 5), (1, 8), ..., (8, 5)}
pairs = FOREACH inputDataGrouped GENERATE GeneratePairsUDF(inputData.item) AS pairs_bag;
pairsFlat = FOREACH pairs GENERATE FLATTEN(pairs_bag) AS (item1:int, item2:int);

les alias "inputData" et "inputdatagroupé" sont calculés dans le mapper.

les "paires" et les "paires" du réducteur.

si je change le script en supprimant la ligne avec la commande FLATTEN (pairsFlat = FOREACH pairs GENERATE FLATTEN (pairs_bag) AS (item1:int, item2:int);) alors l'exécution donne 5 réducteurs (et ainsi, dans une exécution parallèle).

il semble que la commande FLATTEN soit le problème et évite que beaucoup de réducteurs soient créés.

comment atteindre le même résultat de FLATTEN mais avoir le script exécuté en parallèle (avec de nombreux réducteurs)?

Edit:

expliquer le plan lorsque vous avez deux FOREACH (comme ci-dessus):

Map Plan
inputDataGrouped: Local Rearrange[tuple]{chararray}(false) - scope-32
|   |
|   Project[chararray][0] - scope-33
|
|---inputData: New For Each(false,false)[bag] - scope-29
    |   |
    |   Cast[chararray] - scope-24
    |   |
    |   |---Project[bytearray][0] - scope-23
    |   |
    |   Cast[int] - scope-27
    |   |
    |   |---Project[bytearray][1] - scope-26
    |
    |---inputData: Load(file:///input_data.txt:org.apache.pig.builtin.PigStorage) - scope-22--------


Reduce Plan
pairsFlat: Store(fakefile:org.apache.pig.builtin.PigStorage) - scope-42
|
|---pairsFlat: New For Each(true)[bag] - scope-41
    |   |
    |   Project[bag][0] - scope-39
    |
    |---pairs: New For Each(false)[bag] - scope-38
        |   |
        |   POUserFunc(GeneratePairsUDF)[bag] - scope-36
        |   |
        |   |---Project[bag][1] - scope-35
        |       |
        |       |---Project[bag][1] - scope-34
        |
        |---inputDataGrouped: Package[tuple]{chararray} - scope-31--------
Global sort: false

expliquer le plan lorsqu'il n'y a QU'un seul avant avec un emballage aplati de L'UDF:

Map Plan
inputDataGrouped: Local Rearrange[tuple]{chararray}(false) - scope-29
|   |
|   Project[chararray][0] - scope-30
|
|---inputData: New For Each(false,false)[bag] - scope-26
    |   |
    |   Cast[chararray] - scope-21
    |   |
    |   |---Project[bytearray][0] - scope-20
    |   |
    |   Cast[int] - scope-24
    |   |
    |   |---Project[bytearray][1] - scope-23
    |
    |---inputData: Load(file:///input_data.txt:org.apache.pig.builtin.PigStorage) - scope-19--------


Reduce Plan
pairs: Store(fakefile:org.apache.pig.builtin.PigStorage) - scope-36
|
|---pairs: New For Each(true)[bag] - scope-35
    |   |
    |   POUserFunc(GeneratePairsUDF)[bag] - scope-33
    |   |
    |   |---Project[bag][1] - scope-32
    |       |
    |       |---Project[bag][1] - scope-31
    |
    |---inputDataGrouped: Package[tuple]{chararray} - scope-28--------
Global sort: false