setup 函数原码:(摘自《hadoop实战》)
*Called once at the start of the task. protected void setup(Context context) throws IOException,InterruptedException{}从注释可得知,setup函数在Task启动时就调用。 在MapReduce中作业会被组织成MapTask和ReduceTask。 每个Task都以Map类或Reduce类为处理方法主体, 输入分片为处理方法的输入,自己的分片处理完后Task就销毁了。 从这里看出,setup函数在task启动后数据处理前就调用一次 而覆盖的Map函数和Reduce函数会针对输入分片的每个Key调用一次, 所以setup函数可以看作Task上一个全局处理。 利用setup函数的特性,可以将Map或Reduce函数中的的重复处理放到setup函数中。 如老师给的Exercise_2中的"name"但需要注意的是,调用setup函数只是对应的Task上全局操作,而不是整个作业的全局操作。可以先用api把本地的文件传到hdfs中的 /user/hadoop/test 里去
//本地文件上传到HDFS上public static void upload(String src,String dst) throws FileNotFoundException,IOException{ InputStream in = new BufferedInputStream(new FileInputStream(src)); //得到配置对象 Configuration conf = new Configuration(); //文件系统 FileSystem fs = FileSystem.get(URI.create(dst), conf); //输出流 OutputStream out = fs.create(new Path(dst), new Progressable() { public void progress() { System.out.println("上传完一个设定缓存区大小容量的文件!"); } }); //连接两个流,形成通道,使输入流向输出流传输数据 IOUtils.copyBytes(in, out, 4096,true); } 上传的时候调用这个函数就可以了 例如 upload("/home/jack/test/test.txt","/user/hadoop/test/test");前面的是本地目录中的文件,后面是hdfs中的文件 注意 必须两者都必须是“路径+文件名” 不能没有文件名Configuration conf = new Configuration();
conf.setStrings("job_parms", "aaabbc"); //关键就是这一句 Job job = new Job(conf, "load analysis"); job.setJarByClass(LoadAnalysis.class); job.setMapperClass(LoadMapper.class); job.setReducerClass(LoadIntoHbaseReduce.class); job.setMapOutputKeyClass(Text.class); job.setMapOutputValueClass(Text.class); FileInputFormat.addInputPath(job, new Path(otherArgs[0]));@Override
protected void setup(Context context) throws IOException, InterruptedException { try { //从全局配置获取配置参数 Configuration conf = context.getConfiguration(); String parmStr = conf.get("job_parms"); //这样就拿到了 ...... } catch (SQLException e) { e.printStackTrace(); } }全局文件:hadoop有distributed cache来保存全局文件,保证所有node都可以访问,使用类名为DistributedCache