株洲市哪里有做公司官方网站,广州网站营销推广设计,网站模板代理电话,怎么制作自己的水印目录 五、Hadoop 2.0访问接口#xff08;一#xff09;访问接口综述#xff08;二#xff09;浏览器接口#xff08;三#xff09;命令行接口 六、Hadoop 2.0编程接口#xff08;一#xff09;HDFS编程#xff08;二#xff09;Yarn编程 五、Hadoop 2.0访问接口
一访问接口综述二浏览器接口三命令行接口 六、Hadoop 2.0编程接口一HDFS编程二Yarn编程 五、Hadoop 2.0访问接口
一访问接口综述 Hadoop 2.0分为相互独立的几个模块访问各个模块的方式也是相互独立的但每个模块访问方式可分为浏览器接口、Shell接口和编程接口。
二浏览器接口
Web地址配置文件配置参数HDFShttp://NameNodeHostName:50070hdfs-site.xml{dfs.namenode.http-address}Yarnhttp://ResourceManagerHostName:8088yarn-site.xml{yarn.resourcemanager.webapp.address}MapReducehttp://JobHistoryHostName:19888mapred-site.xml{mapreduce.jobhistory.webapp.address} 在Hadoop 2.0里MapReduce是Yarn不可缺少的模块这里的JobHistory是一个任务独立模块用来查看历史任务和MapReduce并行处理算法无关。
三命令行接口
1. HDFS 以tar包方式部署时其执行方式是HADOOP_HOME/bin/hdfs当以完全模式部署时使用HDFS用户执行hdfs即可。 2. Yarn 以tar包方式部署时其执行方式是HADOOP_HOME/bin/yarn当以完全模式部署时使用Yarn用户执行yarn即可。 每一条命令都包含若干条子命令Yarn的Shell命令也主要分为用户命令和管理员命令。
3. Hadoop 以tar包方式部署时其执行方式是HADOOP_HOME/bin/Hadoop当以完全模式部署时在终端直接执行hadoop。 这个脚本既包含HDFS里最常用命令fs即HDFS里的dfs又包含Yarn里最常用命令jar可以说是HDFS和Yarn的结合体。此外distcp用mapreduce来实现两个Hadoop集群之间大规模数据复制。
4. 其他常用命令 sbin/目录下的脚本主要分为两种类型启停服务脚本和管理服务脚本。其中脚本hadoop-daemon.sh可单独用于启动本机服务方便本机调试start/stop类脚本适用于管理整个集群读者只要在命令行下直接使用这些脚本它会自动提示使用方法。 六、Hadoop 2.0编程接口
一HDFS编程 1. HDFS编程实例
【例1】 请编写一简单程序要求实现在HDFS里新建文件myfile并且写入内容“china cstor cstor cstor china”。
代码如下
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();}
}假定程序打包后称为hdfsOperate.jar并假定以joe用户执行程序主类为Write主类前为包名则命令执行如下
[joecMaster~]$ hadoop jar hdfsOperate.jar cn.cstor.data.hadoop.hdfs.write.Write成功执行上述命令后可使用如下两种方式确认文件已经写入HDFS。 第一种方式使用Shell接口以joe用户执行如下命令
[joecMaster~]$ hdfs dfs -cat ls #类似于Linux的ls,列举HDFS文件
[joecMaster~]$ hdfs dfs -cat myfile #类似于Linux的cat,查看文件第二种方式使用Web接口浏览器地址栏打开http://namenodeHostName:50070点击Browse the filesystem进入文件系统接着查看文件/user/jioe/myfile即可。
【例2】 请编写一简单程序要求输出HDFS里刚写入的文件myfile的内容。
代码如下
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();}
}下面是命令执行方式及其结果
[joecMaster~]# hadoop jar hdfsOperate.jar cn.cstor.data.hadoop.hdfs.read.Read
myfile: china cstor cstor china【例3】 请编写一简单代码要求输出HDFS里文件myfile相关属性如文件大小、拥有者、集群副本数最近修改时间等。
代码如下
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());}}
}程序执行方式及其结果如下
[joecMaster~] 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 上面我们通过三个例题介绍了HDFS文件最常用操作但这仅仅是三个小演示程序在真正处理HDFS文件流时可以使用缓冲流将底层文件流一层层包装可大大提高读取效率。
2. HDFS编程基础
1Hadoop统一配置文件类Configuration Hadoop的每一个实体CommonHDFSYarn都有与其相对应的配置文件Configuration类是联系几个配置文件的统一接口。 Hadoop各模块间传递的一切值都必须通过Configuration类实现其他方式均无法获取程序设置的参数若想实现参数最好使用Configuration类的get和set方法。
2取得HDFS文件系统接口 在Hadoop源代码中HDFS相关代码大都存放在org.apache.Hadoop.hdfs包里。但是我们编写代码操作HDFS里的文件时不可以调用这些代码而是通过org.apache.hadoop.fs包里的FileSystem类实现。 FileSystem类是Hadoop访问文件系统的抽象类它不仅可以获取HDFS文件系统服务也可以获取其他文件系统比如本地文件系统服务为程序员访问各类文件系统提供统一接口。
3HDFS常用流和文件状态类 Common还提供了一些处理HDFS文件的常用流fs包下的FSDataInputStreamio包下的缓冲流DataInputBufferutil包下的LineReader等等。用户可以和Java流相互配合使用。
二Yarn编程 Yarn是一个资源管理框架由ResourceManagerRM和NodeManagerNM。但RM和NM不参与计算逻辑。称由ApplicationMaster和Client组成的处理逻辑相同的一类任务为逻辑实体可以定义Map型、MapReduce型、MapReduceMap型和CPU密集型任务。
1. 概念和流程 在资源管理框架中RM负责资源分配NodeManager负责管理本地资源。在计算框架中Client负责提交任务RM启动任务对应的ApplicationMaster。
1编程时使用的协议
① ApplicationClientProtocolClient–ResourceManager。
Client通知RM启动任务如要求RM启动ApplicationMaster获取任务状态或终止任务时使用的协议。
② ApplicationMasterProtocolApplicationMaster–ResourceManager。
ApplicationMaster向RM注册/注销申请资源时用到的协议。
③ ContainerManagerApplicationMaster–NodeManager。
ApplicationMaster启动/停止获取NM上的Container状态信息时所用的协议。
2一个Yarn任务的执行流程简析 Client提交任务时通过调用ApplicationClientProtocol#getNewApplication从RM获取一个ApplicationId然后再通过ApplicationClientProtocol#submitApplication提交任务。 ApplicationMaster则负责此次任务的处理全过程RM会选定一个Container来启动ApplicationMasterApplicationMaster会通过心跳包与RM保持通信ApplicationMaster须向RM注销自己。
3编程步骤小结
① Client端
步骤1获取ApplicationId 步骤2提交任务
② ApplicationMaster端
步骤1注册 步骤2申请资源 步骤3启动Container 步骤4重复步骤2、3直至任务完成 步骤5注销
Yarn提供了三个Application-Master实现DistributedShell、unmanaged-am-launcher、MapReduce。
2. 实例分析 DistributedShell是Yarn自带的一个应用程序编程实例相当于Yarn编程中的“Hello World”它的功能是并行执行用户提交的Shell命令或Shell脚本。 从Hadoop官方网站下载Hadoop-2.2.0-src.tar.gzHadoop源码包并解压后依次进入Hadoop-yarn-project\Hadoop-yarn\Hadoop-yarn-applications下面就是Yarn自带的两个Yarn编程实例。 Client主要向RM提交任务ApplicationMaster向RM申请资源并与NM协商启动Container完成任务。
1Client类主要代码
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); //提交作业从代码中能看到关于RPC的代码已经被上一层代码封装了Client端编程简单地说就是获取YarmClientApplication接着设置ApplicationSubmissionContext最后提交任务。
2ApplicationMaster类最主要代码
//新建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 i0; inumTotalContainers; 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();源码中的ApplicationMaster有1000行上述代码给出了源码里最重要的几个步骤。
3. 代码执行方式
默认情况下Yarn包里已经有分布式Shell的代码了可以使用任何用户执行如下命令
$Hadoop jar /usr/lib/Hadoop-yarn/Hadoop-yarn-applications-distributedshell.jarorg.apache.Hadoop.yarn.applications.distributedshell.Client -jar /usr/lib/Hadoop-yarn/Hadoop-yarn-applications-distributedshell.jar-shell_command /bin/date -num_containers 1004. 实例分析-MapReduce Yarn框架处理MR程序时默认类 InputFormatTextInputFormatRecordReaderLineRecordReaderInputSplitFileSplitMapIdentityMapperCombine不使用PartitionerHashPartitionerGroupCompatator不使用ReduceIdentityReducerOutputFormatFileOutputFormatRecordWriterLineRecordWriterOutputCommitterFileOutputCommitter
MapReduce编程示例——WordCount
下面是MapReduce自带的最简单代码 MapReduce算法实现统计文章中单词出现次数源代码如下
public class WordCount//定义map类,一般继承自Mapper类,里面实现读取单词,写出单词,1public static class TokenizerMapper extends MappercObject, Text, Text, Int Writable {private final static Int Writale one new IntWritable(1);private Text word new Text();//map方法,划分一行文本,读一单词写出一个单词,1public 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 ReducerText, IntWritable, Text, IntWritable {private IntWritable result new IntWritable();public void reduce(Text key, IterableIntWritable values, Context context throws IOException, InterruptedException {int sum 0;for(IntWritable val: values) {sum val.get(); //相当于cstor,1cstor,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,VK类型job.setOutputValueClass(Text.class); //指定Reduce类输出的K,V,V类型job.setOutputFormatClass(TextOutputFormat.class); //指定使用默认输出格式类TextOutputFormat.setOutputPath(job, outputDir); //设置输出结果文件位置System.exit(job.waitForCompletion(true)?0:1); //提交任务并监控任务状态}
}
