Hadoop读取环境变量及setup函数

aa0660

　　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