manupulating itérateur dans mapreduce

Je suis en train d'essayer de trouver la somme des points de l'utilisation d'hadoop, La question que je vais avoir c'est de trouver toutes les valeurs à partir d'une clé dans un seul réducteur. Il ressemble à ceci.

Réducteur:

 public static class Reduce extends MapReduceBase implements
        Reducer<Text, IntWritable, Text, DoubleWritable> {

    public void reduce(Text key, Iterator<IntWritable> values,
            OutputCollector<Text, DoubleWritable> output, Reporter reporter)
            throws IOException {
        Text word = new Text();

        Iterator<IntWritable> tr = values;
        IntWritable v;
        while (tr.hasNext()) {
             v = tr.next();

            Iterator<IntWritable> td = values;
            while (td.hasNext()) {

                IntWritable u = td.next();
                double sum = u+v;
                word.set( u + " + " + v);
                output.collect(word, new DoubleWritable(sum));
            }
        }
    }
}

Et je suis en train de créer deux copies de l'Itérateur variable de sorte que je peux passer par toutes les valeurs de la deuxième itérateur alors que je reçois une valeur unique de la précédente Itérateur( Deux boucles while ci-dessus), mais les deux itérateurs contenir la même valeur, tout le temps.

Je ne suis pas sûr si c'est la bonne façon de le faire, Toute aide est très apprécié.

Merci,

Tsegay

Je suis aussi en train d'essayer de comprendre un problème similaire. J'ai besoin d'aller sur les enregistrements de deux fois dans la fonction de réduction. Je suis en utilisant hadoop streaming avec python et ne sais pas comment rembobiner l'itérateur pour les enregistrements dans le réducteur.
Les itérateurs, par définition, se déplacent dans une seule direction. Donc, une fois que vous avez fait .next() vous avancé, de toute autre instance de l'itérateur vous a tenu également à rappeler à la valeur suivante. Ce n'est rien de spécifique à Hadoop. Quand vous avez dit "essayer de créer deux copies de l'Itérateur variable", vous, en fait, ne pas créer de copie, ils sont tous les deux le même cas, comme @casper a dit. Cela dit, réduire la carte ne peut pas être une bonne solution pour ce problème, un autre moyen serait d'exécuter les boucles while imbriquées à l'extérieur de M/R une fois que vous écrivez le réducteur de sortie quelque part.

OriginalL'auteur tkt986 | 2010-08-14

29

Les itérateurs dans le réducteur ne sont pas aussi simple que vous pourriez le penser.

Le problème est que le nombre total d'éléments que vous êtes à itérer ne peut pas s'en mémoire. Cela signifie que l'itérateur peut être lu à partir du disque. Si vous avez deux copies indépendantes de l'itérateur, alors vous pouvez avoir l'un d'eux, très loin devant les autres qui implique que les données entre les deux itérateurs point ne peut pas être supprimée.

Pour des raisons de simplicité de mise en œuvre, Hadoop ne supporte pas d'avoir plus d'un itérateur pour la réduire les valeurs.

L'impact pratique de cette est que vous ne pouvez pas passer par le même itérateur deux fois. Ce n'est pas agréable, mais c'est le cas. Si vous devez absolument connaître le nombre d'éléments à tenir dans la mémoire, puis vous pouvez copier tous les éléments dans une liste comme suggéré par MrGomez. Si vous ne savez pas, vous pourriez avoir à utiliser de stockage secondaire.

La meilleure approche consiste à revoir la conception de votre programme de sorte que vous n'avez pas besoin de stockage illimité dans le réducteur. Cela peut devenir un peu compliqué, mais il existe des approches du problème.

Pour votre problème particulier, vous avez une croissance quadratique en la taille de la sortie par rapport à la plus grande de réduire d'entrée de jeu. C'est généralement une très mauvaise idée. Dans la plupart des cas, vous n'avez pas besoin de TOUTES les paires de la plus importante des paires. Si vous pouvez couper l'ensemble des paires d'une certaine façon, alors vous êtes prêt et vous pourriez être en mesure de supprimer l'ensemble des paires de contrainte.

Par exemple, si vous essayez de trouver les 100 paires avec la somme la plus importante pour chacun de réduire, vous pouvez garder une file d'attente de priorité avec les 100 plus grandes entrées vu jusqu'à présent et une file d'attente de priorité avec les 100 plus grandes sommes vu jusqu'à présent. Pour chaque nouvelle entrée, vous pouvez former la somme des 100 plus grands nombres vu jusqu'à présent et essayer de coller ces sommes dans la deuxième file d'attente. Enfin, vous devez vous en tenir à la nouvelle entrée dans la première file d'attente et de l'assiette à la fois les files d'attente à 100 éléments en supprimant les valeurs les plus faibles (si nécessaire). Dans le mode de fermeture de la réduire, vous devriez vider la file d'attente de priorité. Cette approche garantit que vous avez seulement besoin min(n^2, 200), les éléments de stockage, ce qui évite de n^2 problème et évite le double de passer par l'entrée en conservant les 100 plus grands articles, plutôt que de tous les éléments vus.

OriginalL'auteur Ted Dunning
12

Je ne suis pas sûr exactement ce que vous essayez d'accomplir, mais je sais que c'est bien le comportement de Hadoop est Itérateurs est un peu étrange. L'Appel D'Itérateur.next() retournera toujours la MÊME instance de IntWritable, avec le contenu de cette instance remplacé par la valeur suivante. Donc la tenue d'une référence à la IntWritable à travers des appels à Itérateur.next() est presque toujours une erreur. Je crois que ce comportement est par la conception pour réduire le montant de la création de l'objet et GC frais généraux.

Un moyen de contourner ce problème est d'utiliser WritableUtils.clone() de cloner l'instance que vous essayez de préserver à travers les appels à Itérateur.next().

Oui, cela m'est arrivé aujourd'hui. Hourra pour Hadoop itérateurs! Est ce comportement documenté nulle part officiellement (sauf pour les blogs et ici)?

OriginalL'auteur bajafresh4life
2

Pour copier Itérateur, vous ne pouvez pas affecter de l'itérateur à une nouvelle variable. Vous devez "clone" un itérateur à une nouvelle variable de classe iterator.
Lorsque l'itérateur Une assigner une autre variable d'itérateur B, les deux variables de l'itérateur sont fait les mêmes données.

OriginalL'auteur casper

En passant par votre question précédente, vous semblez être coincé sur l'itérateur problème piccolbo décrit. La formulation de votre réducteur indique également que vous avez renoncé à ses algorithmes proposés pour l'approche naïve... qui va fonctionner, quoique réduites.

Me permettre de nettoyer votre code un peu avec ma réponse:

//Making use of Hadoop's Iterable reduce, assuming it's available to you
//
// The method signature is:
//
// protected void reduce(KEYIN key, java.lang.Iterable<VALUEIN> values, 
//  org.apache.hadoop.mapreduce.Reducer<KEYIN,VALUEIN,KEYOUT,VALUEOUT>.Context 
//  context) throws java.io.IOException, java.lang.InterruptedException
//
public void reduce(Text key, Iterable<IntWritable> values, Context context)
        throws IOException, InterruptedException {

    //I assume you declare this here to save on GC
    Text outKey = new Text();
    IntWritable outVal = new IntWritable();

    //Since you've forgone piccolbo's approach, you'll need to maintain the
    //data structure yourself. Since we always walk the list forward and
    //wish to optimize the insertion speed, we use LinkedList. Calls to
    //IntWritable.get() will give us an int, which we then copy into our list.
    LinkedList<Integer> valueList = new LinkedList<Integer>();

    //Here's why we changed the method signature: use of Java's for-each
    for (IntWritable iw: values) {
        valueList.add(iw.get());
    }

    //And from here, we construct each value pair as an O(n^2) operation
    for (Integer i: valueList) {
        for (Integer j: valueList) {
            outKey.set(i + " + " + j);
            outVal.set(i + j);
            context.write(outKey, outVal);
        }
    }

    //Do note: I've also changed your return value from DoubleWritable to
    //IntWritable, since you should always be performing integer operations
    //as defined. If your points are Double, supply DoubleWritable instead.
}

Cela fonctionne, mais il fait plusieurs hypothèses qui limitent les performances lors de la construction de votre matrice de distance, y compris en exigeant de la combinaison pour être effectuée dans un seul de réduire fonctionnement.

Envisager piccolbo de l'approche si vous connaissez la taille et la dimension de votre jeu de données d'entrée à l'avance. Il devrait être disponible, dans le pire des cas, en marchant les lignes de l'entrée dans le temps linéaire.

(Voir ce fil pour lesquelles nous ne pouvons pas mettre en œuvre la présente comme une itérateur.)

OriginalL'auteur MrGomez

Vous devez vous connecter pour publier un commentaire.