DistributedCache 分布式缓存

分布式缓存DistributedCache

  • 概念：

    所谓的分布式缓存指的是 hadoop框架可以把用户指定的小文件发送到各个maptask运行的机器上，进行缓存，便于maptask读取该文件内容 进行关联查询操作，这就是所谓的map端join。

  • 适用场合：

    • 通常适用于大文件关联小文件，把小文件进行分布式缓存。

  • 举例说明：

    订单数据 商品数据   把商品进行分布式缓存 通过maptask每处理一个订单 关联一次商品数据

  • 如何高效的利用分布式缓存。

  • hadoop DistributedCache可以把小文件分发到每个maptask那里，但是总不能每处理一条记录去读取一次这个缓存的小文件把？

    • 既然文件不大，就可以把小文件加载的maptask运行的内存中，也就是创建数据集合 保存该小文件。

    • 在什么时间把小文件加载到内存中？

    •  protected void setup(Context context) throws IOException, InterruptedException {
   // NOTHING
}
       
       
       
       
       
       
       
       
      1
       protected void setup(Context context) throws IOException, InterruptedException {
      2
         // NOTHING
      3
      }
       
       

      重写父类的setup方法 该方法会在map方法调用之前调用一次，且调用一次，初始化方法。

      在该方法中完成针对分布式小文件的缓存，加载到内存中，便于后续的map方法调用处理。

    • 如何使用分布式缓存
    • 添加缓存文件：
// job.addArchiveToClassPath(archive);缓存 jar 包到 task 运行节点的 classpath 中
// job.addCacheArchive(uri);缓存压缩包到 task 运行节点的工作目录
// job.addFileToClassPath(file);//缓存普通文件到 task 运行节点的 classpath 中
       
       
       
      4
      4
       
       
       
       
       
      1
      添加缓存文件：
      2
      // job.addArchiveToClassPath(archive);缓存 jar 包到 task 运行节点的 classpath 中
      3
      // job.addCacheArchive(uri);缓存压缩包到 task 运行节点的工作目录
      4
      // job.addFileToClassPath(file);//缓存普通文件到 task 运行节点的 classpath 中
       
       
    • 如果涉及大文件也需要进行分布式缓存执行？
      • 调整maptask中默认可以使用内存大小的上限
        mapreduce.map.memory.mb
         
         
         
        1
         
         
         
         
         
        1
        mapreduce.map.memory.mb
         
         
      • 缓存数据到nosql数据库，比如redis mogodb。 内存数据库。

代码举例：

MapSideJoin

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
import java.io.BufferedReader;
import java.io.FileInputStream;
import java.io.IOException;
import java.io.InputStreamReader;
import java.net.URI;
import java.util.HashMap;
import java.util.Map;

public class MapSideJoin {
    static class MapJoinMapper extends Mapper<LongWritable, Text, Text, NullWritable> {
        //用来缓存小文件（商品文件中的数据）
        Map<String, String> produceMap = new HashMap<String,String>();
        Text k = new Text();

        @Override
        protected void setup(Context context)
                throws IOException, InterruptedException {
            //将商品文件中的数据写到缓存中  千万别写成/ pdts.txt否则会提示找不到该文件
            BufferedReader br = new BufferedReader(new InputStreamReader(new FileInputStream("pdts.txt")));
            String line = null;
            while((line=br.readLine())!=null){
                //一行数据格式为P0001,xiaomi（商品id，商品名称）
                String[] fields = line.split(",");
                produceMap.put(fields[0], fields[1]);
            }
        }
        @Override
        protected void map(LongWritable key, Text value, Context context)
                throws IOException, InterruptedException {
            //一行订单数据    格式为 1001,20170710,P0001,1（订单id，创建时间，商品id，购买商品数量）
            String line = value.toString();
            String[] fields = line.split(",");
            //根据订单数据中商品id在缓存中找出来对应商品信息(商品名称)，进行串接
            String productName = produceMap.get(fields[2]);
            k.set(line+"	"+productName);
            context.write(k, NullWritable.get());
        }
    }
    public static void main(String[] args) throws Exception {
        Configuration conf = new Configuration();
        Job job = Job.getInstance(conf);
        //jar包位置
        job.setJarByClass(MapSideJoin.class);

        job.setMapperClass(MapJoinMapper.class);
        job.setMapOutputKeyClass(Text.class);
        job.setMapOutputValueClass(NullWritable.class);
        //设置最终输出类型
        job.setOutputKeyClass(Text.class);
        job.setOutputValueClass(NullWritable.class);
        //指定需要缓存一个文件到所有的maptask运行节点工作目录
//      job.addArchiveToClassPath(archive);缓存jar包到task运行节点的classpath中
//      job.addCacheArchive(uri);缓存压缩包到task运行节点的工作目录
//      job.addFileToClassPath(file);//缓存普通文件到task运行节点的classpath中

        //将产品表文件缓存到task工作节点的工作目录中去
        //缓存普通文件到task运行节点的工作目录(hadoop帮我们完成)
        job.addCacheFile(new URI("/mapjoincache/pdts.txt"));

        //不需要reduce，那么也就没有了shuffle过程
        job.setNumReduceTasks(0);

        FileInputFormat.setInputPaths(job, new Path("/test/mapjoininput/"));
        FileOutputFormat.setOutputPath(job, new Path("/test/mapjoinoutput"));

        boolean ex = job.waitForCompletion(true);
        System.exit(ex?0:1);
    }
}
 
 
 
x
 
 
 
 
 
1
import org.apache.hadoop.conf.Configuration;
2
import org.apache.hadoop.fs.Path;
3
import org.apache.hadoop.io.LongWritable;
4
import org.apache.hadoop.io.NullWritable;
5
import org.apache.hadoop.io.Text;
6
import org.apache.hadoop.mapreduce.Job;
7
import org.apache.hadoop.mapreduce.Mapper;
8
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
9
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
10
import java.io.BufferedReader;
11
import java.io.FileInputStream;
12
import java.io.IOException;
13
import java.io.InputStreamReader;
14
import java.net.URI;
15
import java.util.HashMap;
16
import java.util.Map;
17








18




public class MapSideJoin {








19




    static class MapJoinMapper extends Mapper<LongWritable, Text, Text, NullWritable> {








20




        //用来缓存小文件（商品文件中的数据）








21




        Map<String, String> produceMap = new HashMap<String,String>();








22




        Text k = new Text();








23












24




        @Override








25




        protected void setup(Context context)








26




                throws IOException, InterruptedException {








27




            //将商品文件中的数据写到缓存中  千万别写成/ pdts.txt否则会提示找不到该文件








28




            BufferedReader br = new BufferedReader(new InputStreamReader(new FileInputStream("pdts.txt")));








29




            String line = null;








30




            while((line=br.readLine())!=null){








31




                //一行数据格式为P0001,xiaomi（商品id，商品名称）








32




                String[] fields = line.split(",");








33




                produceMap.put(fields[0], fields[1]);








34




            }








35




        }








36




        @Override








37




        protected void map(LongWritable key, Text value, Context context)








38




                throws IOException, InterruptedException {








39




            //一行订单数据    格式为 1001,20170710,P0001,1（订单id，创建时间，商品id，购买商品数量）








40




            String line = value.toString();








41




            String[] fields = line.split(",");








42




            //根据订单数据中商品id在缓存中找出来对应商品信息(商品名称)，进行串接








43




            String productName = produceMap.get(fields[2]);








44




            k.set(line+"	"+productName);








45




            context.write(k, NullWritable.get());








46




        }








47




    }








48




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








49




        Configuration conf = new Configuration();








50




        Job job = Job.getInstance(conf);








51




        //jar包位置








52




        job.setJarByClass(MapSideJoin.class);








53












54




        job.setMapperClass(MapJoinMapper.class);








55




        job.setMapOutputKeyClass(Text.class);








56




        job.setMapOutputValueClass(NullWritable.class);








57




        //设置最终输出类型








58




        job.setOutputKeyClass(Text.class);








59




        job.setOutputValueClass(NullWritable.class);








60




        //指定需要缓存一个文件到所有的maptask运行节点工作目录








61




//      job.addArchiveToClassPath(archive);缓存jar包到task运行节点的classpath中








62




//      job.addCacheArchive(uri);缓存压缩包到task运行节点的工作目录








63




//      job.addFileToClassPath(file);//缓存普通文件到task运行节点的classpath中








64












65




        //将产品表文件缓存到task工作节点的工作目录中去








66




        //缓存普通文件到task运行节点的工作目录(hadoop帮我们完成)








67




        job.addCacheFile(new URI("/mapjoincache/pdts.txt"));








68












69




        //不需要reduce，那么也就没有了shuffle过程








70




        job.setNumReduceTasks(0);








71












72




        FileInputFormat.setInputPaths(job, new Path("/test/mapjoininput/"));








73




        FileOutputFormat.setOutputPath(job, new Path("/test/mapjoinoutput"));








74












75




        boolean ex = job.waitForCompletion(true);








76




        System.exit(ex?0:1);








77




    }








78




} 








79


















 


 










 







        

	      

	    

	  

	  

	    
      
        【推广】
        免费学中医，健康全家人
      
	  

    

    

          原文地址：https://www.cnblogs.com/TiePiHeTao/p/65e8c97e244c4e0d9a2ad2a607f51f41.html