Как поместить файлы в память с помощью распределенного кеша Hadoop?

Question

Насколько я знаю, распределенный кеш копирует файлы на каждый узел, а затем отображает или уменьшает чтение файлов из локальной файловой системы.

Мой вопрос: есть ли способ разместить наши файлы в памяти с использованием распределенного кеша Hadoop, чтобы каждая карта или сокращение могли читать файлы непосредственно из памяти?

Программа MapReduce распространяет png-изображение, которое составляет около 1M для каждого узла, затем каждая задача карты считывает изображение из распределенного кеша и выполняет некоторую обработку изображения с другого изображения с ввода карты.

Answer 1

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); 
    } 

} 

Answer 2

большой вопрос. Я также пытаюсь решить подобную проблему. Я не думаю, что Hadoop поддерживает в кеше памяти из коробки. Однако для этой цели не должно быть очень сложно иметь другую в кеше памяти где-то в сетке. Мы можем передать местоположение кеша и имя параметра как часть конфигурации задания.

Что касается вышеприведенного примера кода, он не отвечает на исходный вопрос. Кроме того, он демонстрирует не оптимальный образец кода. В идеале вы должны получить доступ к файлу кэша как часть метода setup() и кэшировать любую информацию, которую вы можете использовать, как часть метода map(). В приведенном выше примере кеш-файл будет читаться один раз для каждой пары ключ-значение, которая ухудшается с выполнением задания mapreduce.