Apache Pig: APLATIR et de l'exécution parallèle de réducteurs

Question

J'ai mis en place un Apache script Pig. Lorsque j'exécute le script, il en résulte dans de nombreux cartographes pour une étape spécifique, mais n'a qu'un seul réducteur pour cette étape. En raison de cette condition (de nombreux cartographes, un réducteur) le cluster Hadoop est quasiment inactif alors que le seul réducteur exécute. Afin de mieux utiliser les ressources de la grappe, j'aimerais aussi avoir de nombreux réducteurs qui s'exécutent en parallèle.

Même si j'ai mis le parallélisme dans le Cochon script à l'aide de l'ENSEMBLE 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);

Le "inputData" et "inputDataGrouped' alias sont calculées dans le mappeur.

Les "paires" et "pairsFlat" dans le réducteur.

Si j'ai modifier le script en retrait de la ligne avec la mise à plat de commande (pairsFlat = FOREACH paires de GÉNÉRER APLATIR(pairs_bag) COMME (item1:int, item2:int);) alors les résultats de l'exécution dans 5 réducteurs (et donc dans une exécution parallèle).

Il semble que la mise à plat de commande est le problème et évite que de nombreuses réductions sont créés.

Quelqu'un sait comment je pourrais atteindre le même résultat de l'APLATIR, mais ayant le script en cours d'exécution en parallèle (avec de nombreux réducteurs)?

Merci beaucoup d'avance pour votre soutien!

Meilleures salutations, Christian

Edit:

EXPLIQUER le plan, une fois avoir 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

EXPLIQUEZ-plan lors de n'avoir qu'un FOREACH avec APLATIR l'habillage 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

Answer 1

Il n'y a aucune garantie si le cochon utilise la configuration DEFAULT_PARALLEL de la valeur pour toutes les étapes du script pig. Essayez PARALLÈLE avec votre join/groupe de l'étape où vous vous sentez en prenant le temps (Dans votre cas, le GROUPE de l'étape).

 inputDataGrouped = GROUP inputData BY (group_name) PARALLEL 67;

Si encore il n'est pas de travail, alors vous pourriez avoir à voir vos données pour l'asymétrie question.

Answer 2

Je pense qu'il y a une asymétrie dans les données. Seul un petit nombre de contributeurs sont la production exponentielle de grande puissance. Regarder la distribution de clés dans vos données. Comme les données ne contient que peu de Groupes avec un grand nombre de dossiers.

Answer 3

J'ai essayé de "définir par défaut parallèle" et "PARALLÈLE 100", mais pas de chance. Cochon utilise encore 1 réducteur.

Il s'est avéré que j'ai pour générer un nombre aléatoire de 1 à 100 pour chaque enregistrement et de groupe de ces dossiers par ce nombre aléatoire.

Nous sommes une perte de temps de regroupement, mais il est beaucoup plus rapide pour moi parce que maintenant je peux utiliser plus réducteurs.

Voici le code (AUTEUR est mon propre UDF):

tmpRecord = FOREACH record GENERATE (int)(RANDOM()*100.0) as rnd, data;
groupTmpRecord = GROUP tmpRecord BY rnd;
result = FOREACH groupTmpRecord GENERATE FLATTEN(SUBMITTER(tmpRecord));