Fractionnement de la sortie du réducteur dans Hadoop

Question

Les fichiers de sortie produits par mon opération de réduction sont énormes (1 Go après la compression). Je veux qu'il soit divisé en petits fichiers de 200 Mo. Existe-t-il une propriété ou une classe Java permettant de diviser le résultat de la réduction par taille ou par nombre de lignes ? Je ne peux pas augmenter le nombre de réducteurs car cela a un impact négatif sur les performances du job hadoop.

Answer 1

Je suis curieux de savoir pourquoi vous ne pouvez pas simplement utiliser plus de réducteurs, mais je vous prends au mot.

Une option que vous pouvez faire est d'utiliser MultipleOutputs et d'écrire dans plusieurs fichiers à partir d'un réducteur. Par exemple, disons que le fichier de sortie de chaque réducteur est de 1 Go et que vous voulez des fichiers de 256 Mo à la place. Cela signifie que vous devez écrire 4 fichiers par réducteur plutôt qu'un seul fichier.

Dans votre pilote d'emploi, faites ceci :

JobConf conf = ...;

// You should probably pass this in as parameter rather than hardcoding 4.
conf.setInt("outputs.per.reducer", 4);

// This sets up the infrastructure to write multiple files per reducer.
MultipleOutputs.addMultiNamedOutput(conf, "multi", YourOutputFormat.class, YourKey.class, YourValue.class);

Dans votre réducteur, faites ceci :

@Override
public void configure(JobConf conf) {
  numFiles = conf.getInt("outputs.per.reducer", 1);
  multipleOutputs = new MultipleOutputs(conf);

  // other init stuff
  ...
}

@Override
public void reduce(YourKey key
                   Iterator<YourValue> valuesIter,
                   OutputCollector<OutKey, OutVal> ignoreThis,
                   Reporter reporter) {
    // Do your business logic just as you're doing currently.
    OutKey outputKey = ...;
    OutVal outputVal = ...;

    // Now this is where it gets interesting. Hash the value to find
    // which output file the data should be written to. Don't use the
    // key since all the data will be written to one file if the number
    // of reducers is a multiple of numFiles.
    int fileIndex = (outputVal.hashCode() & Integer.MAX_VALUE) % numFiles;

    // Now use multiple outputs to actually write the data.
    // This will create output files named: multi_0-r-00000, multi_1-r-00000,
    // multi_2-r-00000, multi_3-r-00000 for reducer 0. For reducer 1, the files
    // will be multi_0-r-00001, multi_1-r-00001, multi_2-r-00001, multi_3-r-00001.
    multipleOutputs.getCollector("multi", Integer.toString(fileIndex), reporter)
      .collect(outputKey, outputValue);
}

@Overrider
public void close() {
   // You must do this!!!!
   multipleOutputs.close();
}

Ce pseudo code a été écrit avec l'ancienne api mapreduce à l'esprit. Des apis équivalentes existent cependant en utilisant l'api mapreduce, donc dans tous les cas, vous devriez être prêts.

Answer 2

Il n'y a pas de propriété pour faire ça. Vous devrez écrire votre propre format de sortie et votre propre enregistreur.