YARN集群的mapreduce测试（六）

两张表链接操作（分布式缓存）：

----------------------------------
假设：
其中一张A表，只有20条数据记录（比如group表）
另外一张非常大，上亿的记录数量（比如user表）
----------------------------------
策略：
将数据集小的文件直接装载到内存，然后迭代大文件记录；

分布式缓存的两种角度理解（针对较小数据集）：
1、将文件不切块，直接存储到各个节点上的本地磁盘中，这种模式的缓存只是减少了网络IO，磁盘IO并没有减少；
2、将文件不切块，直接存储到各个节点上的任务进程内存中（MapTask进程，JVM会开辟一块任务内存），这种模式的缓存从根本上取消了该文件的读取IO操作；

测试准备：

首先同步时间，然后master先开启hdfs集群，再开启yarn集群；用jps查看：

master上： 先有NameNode、SecondaryNameNode;再有ResourceManager;

slave上：   先有DataNode；再有NodeManager；

如果master启动hdfs和yarn成功，但是slave节点有的不成功，则可以使用如下命令手动启动： 

hadoop-daemon.sh start datanode

yarn-daemon.sh start nodemanager

然后在集群中的“/data/cacheusergroup/smalldata/”目录下上传本地文件group表，在“/data/cacheusergroup/src”目录下长传本地文件user表；内容如下：

测试目标：

1、将文件不切块，直接存储到各个节点上的本地磁盘中，这种模式的缓存只是减少了网络IO，磁盘IO并没有减少；
2、将文件不切块，直接存储到各个节点上的任务进程内存中（MapTask进程，JVM会开辟一块任务内存），这种模式的缓存从根本上取消了该文件的读取IO操作；

测试代码：大数据学习交流QQ群：217770236 让我们一起学习大数据

1、直接存储到各个节点上的本地磁盘中：

在mapper中的setup中要获取到数据在本地磁盘中的路径：

（从本地磁盘读取数据到内存中缓存起来，之后在map方法的迭代中就可以直接使用）

//因为StringBuilder比StringBuffer效率更高;StringBuilder是多线程的，当不存在内容修改时，是最好的选择；

package com.mmzs.bigdata.yarn.mapreduce;

import java.io.BufferedReader;
import java.io.File;
import java.io.FileInputStream;
import java.io.IOException;
import java.io.InputStreamReader;
import java.net.URI;
import java.util.HashMap;
import java.util.Map;

import org.apache.commons.io.IOUtils;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.MRJobConfig;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.util.StringUtils;

public class CacheGroupMapper extends Mapper<LongWritable, Text, Text, Text> {
    
    private Text outKey;
    private Text outValue;
    private Map<String, String> smallMap;
    
    @Override
    protected void setup(Mapper<LongWritable, Text, Text, Text>.Context context)
            throws IOException, InterruptedException {
        outKey = new Text();
        outValue = new Text();
        smallMap = new HashMap<String, String>();
        
        Configuration conf = context.getConfiguration();
                
        //从本地磁盘读取数据到内存中缓存起来，之后在map方法的迭代中就可以直接使用
        //获取缓存到本地磁盘的路径
//        /*
        String[] localPaths = conf.getStrings(MRJobConfig.CACHE_LOCALFILES);
        Path[] path = StringUtils.stringToPath(localPaths);
        Path localPath = path[0];
        URI localUri = localPath.toUri();
        File fp = new File(localUri);
        BufferedReader br =null;
        String line = null; 
        try{
            //BufferedReader类写入文件的过程；读的是group表
            br = new BufferedReader(new InputStreamReader(new FileInputStream(fp)));
            while (null != (line=br.readLine()) ) {
                String[] fields = line.split("\s+");
                smallMap.put(fields[0], fields[1]);
            }
        }finally{
            IOUtils.closeQuietly(br);
        }
//         */
    }

    @Override
    protected void map(LongWritable key, Text value, Mapper<LongWritable, Text, Text, Text>.Context context)
            throws IOException, InterruptedException {
        String line = value.toString();
        String[] fields = line.split("\s+");
        
        String groupId = fields[2];//user表中第三列是groupId
        String groupInfo = smallMap.get(groupId);
        if (null == groupInfo) {
            return;
        }
        
        outKey.set(fields[2]);//这是user表中的第三列groupId
        //因为StringBuilder比StringBuffer效率更高;(因为map是一行一行的读取内容的不存在线程安全问题，所以使用StringBuilder是更好的选择)
        outValue.set(new StringBuilder(groupInfo).append("	").append(fields[0]).append("	").append(fields[1]).toString());
        context.write(outKey, outValue);
    }
    
    @Override
    protected void cleanup(Mapper<LongWritable, Text, Text, Text>.Context context)
            throws IOException, InterruptedException {
        outKey = null;
        outValue = null;
    }
    
}

在reducer中直接将mapper中的输出结果汇总：

package com.mmzs.bigdata.yarn.mapreduce;

import java.io.IOException;
import java.util.Iterator;

import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Reducer;

public class CacheGroupReducer extends Reducer<Text, Text, Text, Text> {
    
    
    @Override
    protected void setup(Reducer<Text, Text, Text, Text>.Context context) throws IOException, InterruptedException {
    }

    @Override
    protected void reduce(Text key, Iterable<Text> values, Reducer<Text, Text, Text, Text>.Context context)
            throws IOException, InterruptedException {
        Iterator<Text> its = values.iterator();
        
        while (its.hasNext()) {
            context.write(key, its.next());
        }
        
    }
    
    @Override
    protected void cleanup(Reducer<Text, Text, Text, Text>.Context context) throws IOException, InterruptedException {
    }


}

在driver中添加如下语句：

//将数据缓存到各个存储节点的本地磁盘中去;这是一种提升冗余度来提升性能的
ugJob.addCacheFile(new URI("hdfs://master01:9000/data/cacheusergroup/smalldata/group"));

package com.mmzs.bigdata.yarn.mapreduce;

import java.io.IOException;
import java.net.URI;
import java.net.URISyntaxException;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;

/**
 * @author hadoop
 *
 */
public class CacheGroupDriver {

    private static FileSystem fs;
    private static Configuration conf;
    static {
        String uri = "hdfs://master01:9000/";
        conf = new Configuration();
        try {
            fs = FileSystem.get(new URI(uri), conf, "hadoop");
        } catch (IOException e) {
            e.printStackTrace();
        } catch (InterruptedException e) {
            e.printStackTrace();
        } catch (URISyntaxException e) {
            e.printStackTrace();
        }
    }
    
    public static void main(String[] args) throws IOException, ClassNotFoundException, InterruptedException, URISyntaxException {
        if (null==args || args.length==0) return;
        Job ugJob = getJob(args);
        if (null == ugJob) {
            return;
        }
        //提交Job到集群并等待Job运行完成，参数true表示将Job运行时的状态信息返回到客户端
        boolean flag = false;
        flag = ugJob.waitForCompletion(true);
        System.exit(flag?0:1);
    }
    
    /**
     * 获取Job实例
     * @param args
     * @return
     * @throws IOException
     * @throws URISyntaxException 
     */
    public static Job getJob(String[] args) throws IOException, URISyntaxException {
        if (null==args || args.length<2) return null;
        //放置需要处理的数据所在的HDFS路径
        Path inputPath = new Path(args[0]);
        //放置Job作业执行完成之后其处理结果的输出路径
        Path outputPath = new Path(args[1]);

        if (fs.exists(outputPath)) {
            fs.delete(outputPath, true);//true表示递归删除
        }
        
        //获取Job实例
        Job ugJob = Job.getInstance(conf, "UserGroupJob01");
        //设置运行此jar包入口类
        //ugJob的入口是WordCountDriver类
        ugJob.setJarByClass(CacheGroupDriver.class);
        //设置Job调用的Mapper类
        ugJob.setMapperClass(CacheGroupMapper.class);
        //设置Job调用的Reducer类（如果一个Job没有Reducer则可以不调用此条语句）
        ugJob.setReducerClass(CacheGroupReducer.class);
        
        //设置MapTask的输出键类型
        ugJob.setMapOutputKeyClass(Text.class);
        //设置MapTask的输出值类型
        ugJob.setMapOutputValueClass(Text.class);
        
        //设置整个Job的输出键类型（如果一个Job没有Reducer则可以不调用此条语句）
        ugJob.setOutputKeyClass(Text.class);
        //设置整个Job的输出值类型（如果一个Job没有Reducer则可以不调用此条语句）
        ugJob.setOutputValueClass(Text.class);
        
        //将数据缓存到各个存储节点的本地磁盘中去;
        //这是一种提升冗余度来提升性能的
        ugJob.addCacheFile(new URI("hdfs://master01:9000/data/cacheusergroup/smalldata/group"));
        
        //设置整个Job需要处理数据的输入路径
        FileInputFormat.setInputPaths(ugJob, inputPath);
        //设置整个Job计算结果的输出路径
        FileOutputFormat.setOutputPath(ugJob, outputPath);
        
        return ugJob;
    }
    
}

2、直接存储到各个节点上的任务进程内存中：

（下面是直接从集群路径中读取文件缓存到本地的内存中的操作方式，第一次读取就会从集群中将数据读取，需要克服一次网络IO，然后通过文件系统FileSystem来存到JVM开辟的一块任务内存smallMap中，smallMap本就是存在内存中的，当内存不足时，会发生数据丢失）

在Mapper中的setup中要获取到数据在集群中的路径：

A、主要是修改了Mapper中的setup方法：

    @Override
    protected void setup(Mapper<LongWritable, Text, Text, Text>.Context context)
            throws IOException, InterruptedException {
        outValue = new Text();
        smallMap = new HashMap<String, String>();
        outKey=new Text();
        Configuration conf = context.getConfiguration();
        
        //下面是直接从集群路径中读取文件缓存到本地的内存中的操作方式
        //第一次读取就会从集群中将数据读取，通过文件系统FileSystem来存到内存smallMap中
        //smallMap本就是存在内存中的，当内存不足时，会发生数据丢失；
        URI[] uris = context.getCacheFiles();
        URI uri = uris[0];
        Path clusterPath = new Path(uri);
        FileSystem fs = clusterPath.getFileSystem(conf);
        
        InputStream fis = fs.open(clusterPath);
        String line=null;
        BufferedReader br=null;
        try {
                br=new BufferedReader(new InputStreamReader(fis));
                while(null!=(line=br.readLine())){
                    String[] fields=line.split("\s+");
                    smallMap.put(fields[0], fields[1]);
                }
        }finally{
            if(br!=null) br.close();
        }
｝

Reduce不做修改；Driver类也不做修改；能得到同样的结果；

B、如果在Driver中删除如下语句：

ugJob.addCacheFile(new URI("hdfs://master01:9000/data/cacheusergroup/smalldata/group"));

那么Mapper中的setup方法修改为：

    @Override
    protected void setup(Mapper<LongWritable, Text, Text, Text>.Context context)
            throws IOException, InterruptedException {
        outValue = new Text();
        smallMap = new HashMap<String, String>();
        outKey=new Text();
        Configuration conf = context.getConfiguration();
        
        //下面是直接从集群路径中读取文件缓存到本地的内存中的操作方式
        //第一次读取就会从集群中将数据读取，通过文件系统FileSystem来存到内存smallMap中
        //smallMap本就是存在内存中的，当内存不足时，会发生数据丢失；
        FileSystem fs = null;
        Path clusterPath = null;
        try {
            fs = FileSystem.get(new URI("hdfs://master01:9000/"), conf, "hadoop");
            clusterPath = new Path("hdfs://master01:9000/data/cacheusergroup/smalldata/group");
        } catch (URISyntaxException e) {
            e.printStackTrace();
        }
        
        InputStream fis = fs.open(clusterPath);
        String line=null;
        BufferedReader br=null;
        try {
                br=new BufferedReader(new InputStreamReader(fis));
                while(null!=(line=br.readLine())){
                    String[] fields=line.split("\s+");
                    smallMap.put(fields[0], fields[1]);
                }
        }finally{
            if(br!=null) br.close();
        }
        
    }

package com.mmzs.bigdata.yarn.mapreduce;

import java.io.BufferedReader;
import java.io.File;
import java.io.FileInputStream;
import java.io.IOException;
import java.io.InputStream;
import java.io.InputStreamReader;
import java.net.URI;
import java.net.URISyntaxException;
import java.util.HashMap;
import java.util.Map;

import org.apache.commons.io.IOUtils;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.MRJobConfig;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.util.StringUtils;

public class CacheGroupMapper extends Mapper<LongWritable, Text, Text, Text> {
    
    private Text outKey;
    private Text outValue;
    private Map<String, String> smallMap;
    
    @Override
    protected void setup(Mapper<LongWritable, Text, Text, Text>.Context context)
            throws IOException, InterruptedException {
        outValue = new Text();
        smallMap = new HashMap<String, String>();
        outKey=new Text();
        Configuration conf = context.getConfiguration();
        
        //下面是直接从集群路径中读取文件缓存到本地的内存中的操作方式
        //第一次读取就会从集群中将数据读取，通过文件系统FileSystem来存到内存smallMap中
        //smallMap本就是存在内存中的，当内存不足时，会发生数据丢失；
        FileSystem fs = null;
        Path clusterPath = null;
        try {
            fs = FileSystem.get(new URI("hdfs://master01:9000/"), conf, "hadoop");
            clusterPath = new Path("hdfs://master01:9000/data/cacheusergroup/smalldata/group");
        } catch (URISyntaxException e) {
            e.printStackTrace();
        }
        
//        URI[] uris = context.getCacheFiles();
//        URI uri = uris[0];
//        Path clusterPath = new Path(uri);
//        FileSystem fs = clusterPath.getFileSystem(conf);
        
        InputStream fis = fs.open(clusterPath);
        String line=null;
        BufferedReader br=null;
        try {
                br=new BufferedReader(new InputStreamReader(fis));
                while(null!=(line=br.readLine())){
                    String[] fields=line.split("\s+");
                    smallMap.put(fields[0], fields[1]);
                }
        }finally{
            if(br!=null) br.close();
        }
        
        //从本地磁盘读取数据到内存中缓存起来，之后在map方法的迭代中就可以直接使用
        //获取缓存到本地磁盘的路径
        /*
        String[] localPaths = conf.getStrings(MRJobConfig.CACHE_LOCALFILES);
        Path[] path = StringUtils.stringToPath(localPaths);
        Path localPath = path[0];
        URI localUri = localPath.toUri();
        File fp = new File(localUri);
        BufferedReader br =null;
        String line = null; 
        try{
            //BufferedReader类写入文件的过程；读的是group表
            br = new BufferedReader(new InputStreamReader(new FileInputStream(fp)));
            while (null != (line=br.readLine()) ) {
                String[] fields = line.split("\s+");
                smallMap.put(fields[0], fields[1]);
            }
        }finally{
            IOUtils.closeQuietly(br);
        }
        */
    }

    @Override
    protected void map(LongWritable key, Text value, Mapper<LongWritable, Text, Text, Text>.Context context)
            throws IOException, InterruptedException {
        //如果没有setup中的语句，那么每次map都会克服磁盘IO和网络IO        
        String line = value.toString();
        String[] fields = line.split("\s+");
        
        String groupId = fields[2];//user表中第三列是groupId
        String groupInfo = smallMap.get(groupId);
        if (null == groupInfo) {
            return;
        }
        
        outKey.set(fields[2]);//这是user表中的第三列groupId
        //因为StringBuilder比StringBuffer效率更高;(因为map是一行一行的读取内容的不存在线程安全问题，所以使用StringBuilder是更好的选择)
        outValue.set(new StringBuilder(groupInfo).append("	").append(fields[0]).append("	").append(fields[1]).toString());
        context.write(outKey, outValue);
    }
    
    @Override
    protected void cleanup(Mapper<LongWritable, Text, Text, Text>.Context context)
            throws IOException, InterruptedException {
        outKey = null;
        outValue = null;
    }
    
}

Reduce不做修改；也能得到同样的结果；

 如果没有setup中的语句，那么每次map都会克服磁盘IO和网络IO；性能消耗很大，不利于效率的提升；所以出现了缓存。

测试结果：

运行时传入参数是：

如果在客户端eclipse上运行：传参需要加上集群的master的uri即 hdfs://master01:9000

输入路径参数：  /data/cacheusergroup/src

输出路径参数：  /data/cacheusergroup/dst

 测试过程中，忘记初始化outValue，导致输出没结果，希望道友不要犯类似的小错误。