buaashang 2016-01-25
DistributedCache是Hadoop的一个分布式文件缓存类,使用它有时候能完成一些比较方便的事,DistributedCache第一个比较方便的作用就是来完成分布式文件共享这件事,第二个比较有用的场景,就是在执行一些join操作时,将小表放入cache中,来提高连接效率。
那么,散仙今天要介绍的是如何使用DistributedCache来共享全局的缓存文件。
下面我们先通过一个表格来看下,在hadoop中,使用全局变量或全局文件共享的几种方法
序号 | 方法 |
1 | 使用Configuration的set方法,只适合数据内容比较小的场景 |
2 | 将共享文件放在HDFS上,每次都去读取,效率比较低 |
3 | 将共享文件放在DistributedCache里,在setup初始化一次后,即可多次使用,缺点是不支持修改操作,仅能读取 |
本篇散仙,将重点介绍,使用DistributedCache的方法,来共享一些全局配置文件,或变量,通过DistributedCache的addCacheFile方法,我们把HDFS上的一些文件,在hadoop任务启动时,就载入缓存里面,以供全局使用,使用这个方法时,我们需要注意几点,首先我们的本地文件,需要上传到HDFS上,然后再这个方法里,写入加载路径,接下来,我们就可以,在setup初始化时,读取出,其内容,然后在map或reduce方法,执行时,就可以实时的使用这个文件的一些内容了。
散仙,测试共享的一个文件内容如下:
代码如下,注意散仙在setup方法里,读取了文件内容,并打印:
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); } }
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); } }
在web页面上,查询日志输入情况,如下截图所示
当然,我们也可以根据web上的路径,到对应的日志目录下,查找日志的内容,看到上图就说明,我们上传的共享文件,读取成功了,只要在setup方法里面进行初始化后,对我们的程序来说,就是全局共享了,然后我们就可以结合我们的业务逻辑,来处理一些事了。
最后,在简单总结一下:DistributedCache文件共享的模式,只能在集群的环境中使用,如果在Local模式下测试,就会报如下的文件找不到异常:
运行模式: 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
运行模式: 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
如果你很幸运,在1.x的hadoop里看到如下所示的异常,那么你应该考虑如下的几个问题,第一,是不是以Local模式启动的MR任务,第二读取时的路径是不是有问题,使用DistributedCache共享的文件,会在我们每个节点上配置的目录里面找到对应的共享文件:
<?xml version="1.0"?> <?xml-stylesheet type="text/xsl" href="configuration.xsl"?> <!-- Put site-specific property overrides in this file. --> <configuration> <!-- jobtracker的master地址--> <property> <name>mapred.job.tracker</name> <value>192.168.75.130:9001</value> </property> <property> <!-- hadoop的日志输出指定目录--> <name>mapred.local.dir</name> <value>/root/hadoop1.2/mylogs</value> </property> </configuration>
共享的文件,会被下载到每个节点上的指定的文件夹里找到。
散仙找的一个的路径:
/root/hadoop1.2/mylogs/taskTracker/distcache/2726204645197711229_1788685676_88844454/192.168.75.130/root/input
其他的节点上也一样,只不过IP地址不一样,截图如下:
至此,我们就可以使用轻松的来使用DistributedCache来共享一些比较大的文件,或压缩包了。