Comment mettre les fichiers dans la mémoire à l'aide de cache Distribué Hadoop?

Autant que je sache, cache distribué des copies de fichiers à chaque nœud, puis la carte ou réduire lit les fichiers à partir du système de fichiers local.

Ma question est: Est-il un moyen que nous pouvons mettre nos fichiers en mémoire à l'aide de cache distribué Hadoop, de sorte que chaque carte ou de réduire les permet de lire des fichiers directement à partir de la mémoire?

Mon MapReduce programme distribue une image png qui est d'environ 1M à chaque nœud, chaque carte tâche lectures de l'image à partir du cache distribué et fait un peu de traitement de l'image avec une autre photo de l'entrée de la carte.

OriginalL'auteur hequn8128 | 2013-12-12

  1. 2
    import java.io.BufferedReader;
import java.io.FileReader;
import java.io.IOException;
import java.net.URI;
import java.util.StringTokenizer;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.filecache.DistributedCache;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.Reducer;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
import org.apache.hadoop.util.GenericOptionsParser;

public class WordCount {

  public static class TokenizerMapper 
       extends Mapper<Object, Text, Text, IntWritable>{

    private final static IntWritable one = new IntWritable(1);
    private Text word = new Text();

    public void map(Object key, Text value, Context context
                    ) throws IOException, InterruptedException {

          Path[] uris = DistributedCache.getLocalCacheFiles(context
                    .getConfiguration());





                    try{
                        BufferedReader readBuffer1 = new BufferedReader(new FileReader(uris[0].toString()));
                        String line;
                        while ((line=readBuffer1.readLine())!=null){
                            System.out.println(line);

                        }
                        readBuffer1.close(); 
                    }       
                    catch (Exception e){
                        System.out.println(e.toString());
                    }

                  StringTokenizer itr = new StringTokenizer(value.toString());

      while (itr.hasMoreTokens()) {
        word.set(itr.nextToken());
        context.write(word, one);
      }
    }
  }

  public static class IntSumReducer 
       extends Reducer<Text,IntWritable,Text,IntWritable> {
    private IntWritable result = new IntWritable();

    public void reduce(Text key, Iterable<IntWritable> values, 
                       Context context
                       ) throws IOException, InterruptedException {
      int sum = 0;
      for (IntWritable val : values) {
        sum += val.get();
      }
      int length=key.getLength();
      System.out.println("length"+length);
      result.set(sum);
/*      key.set("lenght"+lenght);*/
      context.write(key, result);


    }
  }

  public static void main(String[] args) throws Exception {

      final String NAME_NODE = "hdfs://localhost:9000";
    Configuration conf = new Configuration();

    String[] otherArgs = new GenericOptionsParser(conf, args).getRemainingArgs();
    if (otherArgs.length != 2) {
      System.err.println("Usage: wordcount <in> <out>");
      System.exit(2);
    }
    Job job = new Job(conf, "word count");
    job.setJarByClass(WordCount.class);
    job.setMapperClass(TokenizerMapper.class);
    job.setCombinerClass(IntSumReducer.class);
    job.setReducerClass(IntSumReducer.class);
    job.setOutputKeyClass(Text.class);
    job.setOutputValueClass(IntWritable.class);


    DistributedCache.addCacheFile(new URI(NAME_NODE
      + "/dataset1.txt"),
      job.getConfiguration());



    FileInputFormat.addInputPath(job, new Path(otherArgs[0]));
    FileOutputFormat.setOutputPath(job, new Path(otherArgs[1]));
    System.exit(job.waitForCompletion(true) ? 0 : 1);
  }

}
    Merci. Je sais comment les utiliser de cache distribué. Ma question est de savoir comment mettre des fichiers dans la mémoire du système de fichiers local. Dans votre programme, chaque carte va lire le dataset1.txt du système de fichiers local. Il semble que l'Étincelle peut répondre à ma demande.
    Charger l'image dans setup().

    OriginalL'auteur user.

  2. 0

    grande question. Je suis aussi en train d'essayer de résoudre le problème similaire. Je ne pense pas que Hadoop prend en charge dans la mémoire cache de la boîte. Toutefois, il ne devrait pas être très difficile d'avoir un autre dans la mémoire cache quelque part sur la grille à cet effet. Nous ne pouvons passer à l'emplacement de la cache et le nom du paramètre dans la Configuration du Travail.

    Aussi loin que l'exemple de code ci-dessus, il ne répond pas à la question d'origine. En outre, il met en valeur non-optimale de l'exemple de code. Idéalement, vous devriez accéder au fichier de cache dans le cadre de la méthode setup() et cache tout ce que vous voulez utiliser dans le cadre de la méthode map (). Dans l'exemple ci-dessus le cache de fichier sera lu une seule fois pour chaque paire clé-valeur qui compromet avec la performance de la tâche de mapreduce.

    OriginalL'auteur Saket