如何使用Hadoop的DistributedCache

男娘i 2022-04-21 17:04 163阅读 0赞

\[b\]\[color=green\]\[size=large\]DistributedCache是Hadoop的一个分布式文件缓存类，使用它有时候能完成一些比较方便的事，DistributedCache第一个比较方便的作用就是来完成分布式文件共享这件事，第二个比较有用的场景，就是在执行一些join操作时，将小表放入cache中，来提高连接效率。  
  
  
那么，散仙今天要介绍的是如何使用DistributedCache来共享全局的缓存文件。  
  
下面我们先通过一个表格来看下，在hadoop中，使用全局变量或全局文件共享的几种方法  
\[table\]  
|序号|方法|  
|1|使用Configuration的set方法，只适合数据内容比较小的场景|  
|2|将共享文件放在HDFS上，每次都去读取，效率比较低|  
|3|将共享文件放在DistributedCache里，在setup初始化一次后，即可多次使用，缺点是不支持修改操作，仅能读取|  
\[/table\]  
  
  
本篇散仙，将重点介绍，使用DistributedCache的方法，来共享一些全局配置文件，或变量，通过DistributedCache的addCacheFile方法，我们把HDFS上的一些文件，在hadoop任务启动时，就载入缓存里面，以供全局使用，使用这个方法时，我们需要注意几点，首先我们的本地文件，需要上传到HDFS上，然后再这个方法里，写入加载路径，接下来，我们就可以，在setup初始化时，读取出，其内容，然后在map或reduce方法，执行时，就可以实时的使用这个文件的一些内容了。  
  
散仙，测试共享的一个文件内容如下：  
\[/size\]\[/color\]\[/b\]  
  
\[img\]http://dl2.iteye.com/upload/attachment/0095/4829/ae3d6970-77db-34ff-8ac0-af517fc3e483.jpg\[/img\]  
\[b\]\[color=green\]代码如下，注意散仙在setup方法里，读取了文件内容，并打印：\[/color\]\[/b\]

package com.qin.testdistributed;
    
    import java.io.File;
    import java.io.FileReader;
    import java.io.IOException;
    import java.net.URI;
    import java.util.Scanner;
    
    import org.apache.hadoop.conf.Configuration;
    import org.apache.hadoop.filecache.DistributedCache;
    import org.apache.hadoop.fs.FSDataInputStream;
    import org.apache.hadoop.fs.FileSystem;
    import org.apache.hadoop.fs.Path;
    import org.apache.hadoop.io.IntWritable;
    import org.apache.hadoop.io.LongWritable;
    import org.apache.hadoop.io.Text;
    import org.apache.hadoop.mapred.JobConf;
    import org.apache.hadoop.mapreduce.Job;
    import org.apache.hadoop.mapreduce.Mapper;
    import org.apache.hadoop.mapreduce.Reducer;
    import org.apache.hadoop.mapreduce.lib.db.DBConfiguration;
    import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
    import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
    import org.apache.log4j.pattern.LogEvent;
    
    import org.slf4j.Logger;
    import org.slf4j.LoggerFactory;
    
    import com.qin.operadb.WriteMapDB;
    
    
    /**
     * 测试hadoop的全局共享文件
     * 使用DistributedCached
     * 
     * 大数据技术交流群： 37693216
     * @author qindongliang
     * 
     * ***/
    public class TestDistributed {
       
    
    
    	private static Logger logger=LoggerFactory.getLogger(TestDistributed.class);
    
    
    
    
    
    	private static class FileMapper extends Mapper<LongWritable, Text, Text, IntWritable>{
       
    
    	     	Path path[]=null;
    
    		/**
    		 * Map函数前调用
    		 * 
    		 * */
    		@Override
    		protected void setup(Context context)
    				throws IOException, InterruptedException {
       
    		  logger.info("开始启动setup了哈哈哈哈");
    		    // System.out.println("运行了.........");
    		  Configuration conf=context.getConfiguration();
    		   path=DistributedCache.getLocalCacheFiles(conf);
    	       System.out.println("获取的路径是：  "+path[0].toString());
    	     //  FileSystem fs = FileSystem.get(conf);
    	       FileSystem fsopen= FileSystem.getLocal(conf);
    	     FSDataInputStream in = fsopen.open(path[0]);
    	      // System.out.println(in.readLine());
    //	       for(Path tmpRefPath : path) {
       
    //	           if(tmpRefPath.toString().indexOf("ref.png") != -1) {
       
    //	               in = reffs.open(tmpRefPath);
    //	               break;
    //	           }
    //	       }
    
         // FileReader reader=new FileReader("file://"+path[0].toString());
    //      File f=new File("file://"+path[0].toString());
          // FSDataInputStream in=fs.open(new Path(path[0].toString()));
    	     Scanner scan=new Scanner(in);
    	       while(scan.hasNext()){
       
    	    	   System.out.println(Thread.currentThread().getName()+"扫描的内容:  "+scan.next());
    	       }
    	       scan.close();
    //		
    		// System.out.println("size: "+path.length);
    
    
    		}
    
    
    		@Override
    		protected void map(LongWritable key, Text value,Context context)
    				throws IOException, InterruptedException {
       
    
    		//	System.out.println("map    aaa");
    			//logger.info("Map里的任务");
    			System.out.println("map里输出了");
    		//	logger.info();
    			context.write(new Text(""), new IntWritable(0));
    
    
    		}
    
    
    		 @Override
    		protected void cleanup(Context context)
    				throws IOException, InterruptedException {
       
    
    
    			 logger.info("清空任务了。。。。。。");
    		}
    
    	}
    
    
    	private static class  FileReduce extends Reducer<Object, Object, Object, Object>{
       
    
    
    		@Override
    		protected void reduce(Object arg0, Iterable<Object> arg1,
    				 Context arg2)throws IOException, InterruptedException {
       
    
    
    			System.out.println("我是reduce里面的东西");
    		}
    	}
    
    
    
    	public static void main(String[] args)throws Exception {
       
    
    
    		JobConf conf=new JobConf(TestDistributed.class);
    		//conf.set("mapred.local.dir", "/root/hadoop");
    		 //Configuration conf=new Configuration();
    
    	    // conf.set("mapred.job.tracker","192.168.75.130:9001");
    		//读取person中的数据字段
    	  	   //conf.setJar("tt.jar");
    
    		//注意这行代码放在最前面，进行初始化，否则会报
    		 String inputPath="hdfs://192.168.75.130:9000/root/input";	    
    		 String outputPath="hdfs://192.168.75.130:9000/root/outputsort";
    
    		Job job=new Job(conf, "a");
    		DistributedCache.addCacheFile(new URI("hdfs://192.168.75.130:9000/root/input/f1.txt"), job.getConfiguration());
    		job.setJarByClass(TestDistributed.class);
    		System.out.println("运行模式：  "+conf.get("mapred.job.tracker"));
    		/**设置输出表的的信息  第一个参数是job任务，第二个参数是表名，第三个参数字段项**/
    	   FileSystem fs=FileSystem.get(job.getConfiguration());
    
    		  Path pout=new Path(outputPath);
    		  if(fs.exists(pout)){
       
    			  fs.delete(pout, true);
    			  System.out.println("存在此路径, 已经删除......");
    		  } 
    		 /**设置Map类**/
    		// job.setOutputKeyClass(Text.class);
    		 //job.setOutputKeyClass(IntWritable.class);
    		  job.setMapOutputKeyClass(Text.class);
    		  job.setMapOutputValueClass(IntWritable.class);
    		 job.setMapperClass(FileMapper.class);
    	     job.setReducerClass(FileReduce.class);
    		 FileInputFormat.setInputPaths(job, new Path(inputPath));  //输入路径
             FileOutputFormat.setOutputPath(job, new Path(outputPath));//输出路径  
    
    		System.exit(job.waitForCompletion(true) ? 0 : 1);  
    
    
    
    	}
    
    
    
    
    }

\[b\]\[color=olive\]\[size=large\]在web页面上，查询日志输入情况，如下截图所示\[/size\]\[/color\]\[/b\]  
  
\[img\]http://dl2.iteye.com/upload/attachment/0095/4838/23083f6c-c98d-38b6-8c78-56759aa2b31a.jpg\[/img\]  
\[b\]\[color=olive\]当然，我们也可以根据web上的路径，到对应的日志目录下，查找日志的内容，看到上图就说明，我们上传的共享文件，读取成功了，只要在setup方法里面进行初始化后，对我们的程序来说，就是全局共享了，然后我们就可以结合我们的业务逻辑，来处理一些事了。  
  
  
最后，在简单总结一下：DistributedCache文件共享的模式，只能在集群的环境中使用，如果在Local模式下测试，就会报如下的文件找不到异常：  
\[/color\]\[/b\]

运行模式：  local
    存在此路径, 已经删除......
    WARN - NativeCodeLoader.<clinit>(52) | Unable to load native-hadoop library for your platform... using builtin-java classes where applicable
    WARN - JobClient.copyAndConfigureFiles(746) | Use GenericOptionsParser for parsing the arguments. Applications should implement Tool for the same.
    WARN - JobClient.copyAndConfigureFiles(870) | No job jar file set.  User classes may not be found. See JobConf(Class) or JobConf#setJar(String).
    INFO - FileInputFormat.listStatus(237) | Total input paths to process : 1
    WARN - LoadSnappy.<clinit>(46) | Snappy native library not loaded
    INFO - TrackerDistributedCacheManager.downloadCacheObject(423) | Creating f1.txt in /root/hadoop1.2/hadooptmp/mapred/local/archive/-4778653900406898379_1788685676_88844454/192.168.75.130/root/input-work--1953076903080970848 with rwxr-xr-x
    INFO - TrackerDistributedCacheManager.downloadCacheObject(463) | Cached hdfs://192.168.75.130:9000/root/input/f1.txt as /root/hadoop1.2/hadooptmp/mapred/local/archive/-4778653900406898379_1788685676_88844454/192.168.75.130/root/input/f1.txt
    INFO - TrackerDistributedCacheManager.localizePublicCacheObject(486) | Cached hdfs://192.168.75.130:9000/root/input/f1.txt as /root/hadoop1.2/hadooptmp/mapred/local/archive/-4778653900406898379_1788685676_88844454/192.168.75.130/root/input/f1.txt
    INFO - JobClient.monitorAndPrintJob(1380) | Running job: job_local697121855_0001
    INFO - LocalJobRunner$Job.run(340) | Waiting for map tasks
    INFO - LocalJobRunner$Job$MapTaskRunnable.run(204) | Starting task: attempt_local697121855_0001_m_000000_0
    INFO - Task.initialize(534) |  Using ResourceCalculatorPlugin : null
    INFO - MapTask.runNewMapper(729) | Processing split: hdfs://192.168.75.130:9000/root/input/f1.txt:0+31
    INFO - MapTask$MapOutputBuffer.<init>(949) | io.sort.mb = 100
    INFO - MapTask$MapOutputBuffer.<init>(961) | data buffer = 79691776/99614720
    INFO - MapTask$MapOutputBuffer.<init>(962) | record buffer = 262144/327680
    INFO - TestDistributed$FileMapper.setup(60) | 开始启动setup了哈哈哈哈
    获取的路径是：  /root/hadoop1.2/hadooptmp/mapred/local/archive/-4778653900406898379_1788685676_88844454/192.168.75.130/root/input/f1.txt
    INFO - MapTask$MapOutputBuffer.flush(1289) | Starting flush of map output
    INFO - LocalJobRunner$Job.run(348) | Map task executor complete.
    WARN - LocalJobRunner$Job.run(435) | job_local697121855_0001
    java.lang.Exception: java.io.FileNotFoundException: File /root/hadoop1.2/hadooptmp/mapred/local/archive/-4778653900406898379_1788685676_88844454/192.168.75.130/root/input/f1.txt does not exist.
    	at org.apache.hadoop.mapred.LocalJobRunner$Job.run(LocalJobRunner.java:354)
    Caused by: java.io.FileNotFoundException: File /root/hadoop1.2/hadooptmp/mapred/local/archive/-4778653900406898379_1788685676_88844454/192.168.75.130/root/input/f1.txt does not exist.
    	at org.apache.hadoop.fs.RawLocalFileSystem.getFileStatus(RawLocalFileSystem.java:402)
    	at org.apache.hadoop.fs.FilterFileSystem.getFileStatus(FilterFileSystem.java:255)
    	at org.apache.hadoop.fs.ChecksumFileSystem$ChecksumFSInputChecker.<init>(ChecksumFileSystem.java:125)
    	at org.apache.hadoop.fs.ChecksumFileSystem.open(ChecksumFileSystem.java:283)
    	at org.apache.hadoop.fs.FileSystem.open(FileSystem.java:427)
    	at com.qin.testdistributed.TestDistributed$FileMapper.setup(TestDistributed.java:67)
    	at org.apache.hadoop.mapreduce.Mapper.run(Mapper.java:142)
    	at org.apache.hadoop.mapred.MapTask.runNewMapper(MapTask.java:764)
    	at org.apache.hadoop.mapred.MapTask.run(MapTask.java:364)
    	at org.apache.hadoop.mapred.LocalJobRunner$Job$MapTaskRunnable.run(LocalJobRunner.java:223)
    	at java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:471)
    	at java.util.concurrent.FutureTask$Sync.innerRun(FutureTask.java:334)
    	at java.util.concurrent.FutureTask.run(FutureTask.java:166)
    	at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1110)
    	at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:603)
    	at java.lang.Thread.run(Thread.java:722)
    INFO - JobClient.monitorAndPrintJob(1393) |  map 0% reduce 0%
    INFO - JobClient.monitorAndPrintJob(1448) | Job complete: job_local697121855_0001
    INFO - Counters.log(585) | Counters: 0

\[b\]\[color=green\]\[size=large\]如果你很幸运，在1.x的hadoop里看到如下所示的异常，那么你应该考虑如下的几个问题，第一，是不是以Local模式启动的MR任务，第二读取时的路径是不是有问题，使用DistributedCache共享的文件，会在我们每个节点上配置的目录里面找到对应的共享文件：\[/size\]\[/color\]\[/b\]

<?xml version="1.0"?>
    <?xml-stylesheet type="text/xsl" href="configuration.xsl"?>
    
    
    
    <configuration>
    
    <property> 
    <name>mapred.job.tracker</name> 
    <value>192.168.75.130:9001</value> 
    </property>
    <property>
    
    <name>mapred.local.dir</name>
    <value>/root/hadoop1.2/mylogs</value>
    </property>
    
    </configuration>

\[b\]\[color=green\]\[size=large\]共享的文件，会被下载到每个节点上的指定的文件夹里找到。  
散仙找的一个的路径：  
/root/hadoop1.2/mylogs/taskTracker/distcache/2726204645197711229\_1788685676\_88844454/192.168.75.130/root/input  
其他的节点上也一样，只不过IP地址不一样，截图如下：  
\[/size\]\[/color\]\[/b\]  
  
\[img\]http://dl2.iteye.com/upload/attachment/0095/4846/22727a94-cc34-32a3-a5e0-5631505af567.jpg\[/img\]  
\[b\]\[color=green\]\[size=large\]至此，我们就可以使用轻松的来使用DistributedCache来共享一些比较大的文件，或压缩包了。\[/size\]\[/color\]\[/b\]