DistributedCache 分布式缓存

时间：2019-09-14 21:52:33 阅读：110 评论：0 收藏：0 [点我收藏+]

分布式缓存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 中

 
添加缓存文件：
// job.addArchiveToClassPath(archive);缓存 jar 包到 task 运行节点的 classpath 中
// job.addCacheArchive(uri);缓存压缩包到 task 运行节点的工作目录
// 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+"\t"+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
 
 
 
 
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+"\t"+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);
    }
} 
 
 

DistributedCache 分布式缓存

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

踩

(1)

评论一句话评论（0）

分享档案

更多>

2021年09月23日 (328)
2021年09月24日 (313)
2021年09月17日 (191)
2021年09月15日 (369)
2021年09月16日 (411)
2021年09月13日 (439)
2021年09月11日 (398)
2021年09月12日 (393)
2021年09月10日 (160)
2021年09月08日 (222)

DistributedCache 分布式缓存

分布式缓存DistributedCache

概念：

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

适用场合：

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

举例说明：

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

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

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

如何使用分布式缓存

代码举例：

MapSideJoin

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