Hadoop 2.0：主流开源云架构（四）

public class Write {
    public static void main(String[] args) throws IOException {
        Configuration conf = new Configuration();       //实例化配置文件
        Path inFile = new Path("/user/joe/myfile");      //命名一个文件
        FileSystem hdfs = FileSystem.get(conf);         //获取文件系统
        FSDataOutputStream OutputStream = hdfs.create(inFile);   //获取文件流
        outputStream.writeUTF("china cstor cstor cstor china");   //使用流向文件里写内容
        outputStream.flush();
        outputStream.close();
    }
}

[joe@cMaster~]$ hadoop jar hdfsOperate.jar cn.cstor.data.hadoop.hdfs.write.Write

[joe@cMaster~]$ hdfs dfs -cat ls            #类似于Linux的ls,列举HDFS文件
[joe@cMaster~]$ hdfs dfs -cat myfile        #类似于Linux的cat,查看文件

public class Read {
    public static void main(String[] args) throws IOException {
        Configuration conf = new Configuration();
        Path inFile = new Path("/user/joe/myfile");      //HDFS里欲读取文件的绝对路径
        FileSystem hdfs = FileSystem.get(conf);
        FSDataIutputStream inputStream = hdfs.open(inFile);   //获取输出流
        System.out.println("myfile:"+inputStream.readUTF());   //使用输出流读取文件
        inputStream.close();
    }
}

[joe@cMaster~]# hadoop jar hdfsOperate.jar cn.cstor.data.hadoop.hdfs.read.Read
myfile: china cstor cstor china

public class Status {
    public static void main(String[] args)throws Exception {
        Configuration conf = new Configuration();
        Path file = new Path("/user/joe/myfile");
        System.out.println("FileName:"+file.getName());
        FileSystem hdfs = file.getFileSystem(conf);
        FileStatus[] fileStatus = hdfs.listStatus(file);
        for (FileStatus status: fileStatus) {
            System.out.println("FileOwner:"+status.getOwner());
            System.out.println("FileReplication:"+status.getReplication();
            System.out.println("FileModificationTime:"+new Date(status.getModificationTime());
            System.out.println("FileBlockSize:"+status.getBlockSize());
        }
    }
}

[joe@cMaster~] Hadoop jar hdfsOperate.jar cn.cstor.data.Hadoop.hdfs.file.Status
FileName: myfile
FileOwner: joe
FileReplication: 3
FileModification Time: Tue Nov 12 05:24:02 PST 2013 

YarnClient yarnClient = YarnClient.createYarnClient();    //新建Yarn客户端
yarnClient.start();       //启动Yarn客户端
YarnClientApplication app = yarnClient.createApplication();    //获取提交程序句柄
ApplicationSubmissionContext appContext = app.getApplicationSubmissionContext();   //获取上下文句柄
ApplicationId appId = appContext.getApplicationId();    //获取RM分配的appId 
appContext.setResource(capability);     //设置任务其他信息举例
appContext.setQueue(amQueue);
appContext.setPriority(priority);

//实例化ApplicationMaster对应的Container
ContainerLaunchContext amContainer = Records.newRecord(ContainerLaunchContext.class);
amContainer.setCommands(commands);       //参数commands为用户预执行的Shell命令
appContext.setAMContainerSpec(amContainer);    //指定ApplicationMaster的Container 
yarnClient.submitApplication(appContext);      //提交作业

//新建RM代理
AMRMClientAsync amRMClient = AMRMClientAsync.createAMRMClientAsync(1000, allocListener);
amRMClient.init(conf);
amRMClient.start();
//向RM注册
amRMClient.registerApplicationMaster(appMasterHostname, appMasterRpcPort, appMasterTrackingUrl);
containerListener = createNMCallbackHandler();
//新建NM代理
NMClientAsync nmClientAsync = new NMClientAsyncImpl(containerListener);
nmClientAsync.init(conf);
nmClientAsync.start();
//向RM申请资源
for(int i=0; i<numTotalContainers; ++i) {
    ContainerRequest containerAsk = setupContainerAskForRM();
    amRMClient.addContainerRequest(containerAsk);
}
numRequestedContainers.set(numTotalContainers);
//设置Container上下文
ContainerLaunchContext ctx = Records.newRecord(ContainerLaunchContext.class);
ctx.setCommands(commands);
//要求NM启动Container 
nmClientAsync.startContainerAsync(container, ctx);
//containerListener汇报此NM完成任务后，关闭此NM
nmClientAsync.stop();
//向RM注销
amRMClient.unregisterApplicationMaster(appStatus, appMessage, null);
amRMClient.stop();

$Hadoop jar /usr/lib/Hadoop-yarn/Hadoop-yarn-applications-distributedshell.jar
> org.apache.Hadoop.yarn.applications.distributedshell.Client 
> -jar /usr/lib/Hadoop-yarn/Hadoop-yarn-applications-distributedshell.jar
> -shell_command  '/bin/date' -num_containers 100

public class WordCount
    //定义map类,一般继承自Mapper类,里面实现读取单词,写出<单词,1>
    public static class TokenizerMapper extends Mapperc<Object, Text, Text, Int Writable> {
    private final static Int Writale one = new IntWritable(1);
    private Text word = new Text();
    //map方法,划分一行文本,读一单词写出一个<单词,1>
    public void map(Object key, Text value, Context context)throws IOException, InterruptedException {
        StringTokenizer itr = new StringTokenizer(value.toString());
        while(itr.hasMoreTokens()) {
        word.set(itr.nextToken());
        context.write(word, one);        //写出<单词,1>
    }}} 

//定义reduce类,对相同的单词,把它们<K,VList>中的VList值全部相加
public static class IntSumReducer extends Reducer<Text, IntWritable, Text, IntWritable> {
    private IntWritable result = new IntWritable();
    public void reduce(Text key, Iterable<IntWritable> values, Context context throws IOException, InterruptedException {
        int sum = 0;
        for(IntWritable val: values) {
            sum += val.get();        //相当于<cstor,1><cstor,1>,将两个1相加
        }
        result.set(sum);
        context.write(key,result);      //写出这个单词，和这个单词出现次数<单词,单词出现次数>
    }}
    public static void main(String[] args) throws Exception {    //主方法,函数入口
        Configuration conf = new Configuration();        //实例化配置文件类
        Job job = new Job(conf, "WordCount");        //实例化Job类
        job.setInputFormatClass(TextInputFormat.class);      //指定使用默认输入格式类
        TextInputFormat.setInputPaths(job, inputPaths);       //设置待处理文件的位置
        job.setJarByClass(WordCount.class);        //设置主类名
        job.setMapperClass(TokenizerMapper.class);    //指定使用上述自定义Map类
        job.setMapOutputKeyClass(Text.class);     //指定Map类输出的<K,V>,K类型
        job.setMapOutputValueClass(IntWritable.class);      //指定Map类输出的-K,V>,V类型
        job.setPartitionerClass(HashPartitioner.class);      //指定使用默认的HashPartitioner类
        job.setReducerClass(IntSumReducer.class);     //指定使用上述自定义Reduce类
        job.setNumReduceTasks(Integer.parseInt(numOfReducer);    //指定Reduce个数
        job.setOutputKeyClass(Text.class);        //指定Reduce类输出的<K,V>K类型
        job.setOutputValueClass(Text.class);        //指定Reduce类输出的<K,V>,V类型
        job.setOutputFormatClass(TextOutputFormat.class);      //指定使用默认输出格式类
        TextOutputFormat.setOutputPath(job, outputDir);        //设置输出结果文件位置
        System.exit(job.waitForCompletion(true)?0:1);        //提交任务并监控任务状态
    }
}