HDFS体系结构和Java操作API

1.HDFS体系结构与基本概念

1.1.NameNode

(1)作用

是整个文件系统的管理节点，它维护着整个文件系统的文件目录树，文件/目录的元信息和每个文件对应的数据块列表，接受用户的操作请求。

(2)目录结构

 

在hdfs-default.xml中查看dfs.name.dir、dfs.name.edits.dir对应的代码如下所示：

[html] view plaincopyprint?

1. <</span>property> 
2.   <</span>name>dfs.name.dir</</span>name> 
3.   <</span>value>${hadoop.tmp.dir}/dfs/name</</span>value> 
4.   <</span>description>Determines where on the local filesystem the DFS name node 
5.       should store the name table(fsimage).  If this is a comma-delimited list 
6.       of directories then the name table is replicated in all of the 
7.       directories, for redundancy. </</span>description> 
8. </</span>property> 
9. <</span>property> 
10.   <</span>name>dfs.name.edits.dir</</span>name> 
11.   <</span>value>${dfs.name.dir}</</span>value> 
12.   <</span>description>Determines where on the local filesystem the DFS name node 
13.       should store the transaction (edits) file. If this is a comma-delimited list 
14.       of directories then the transaction file is replicated in all of the   
15.       directories, for redundancy. Default value is same as dfs.name.dir 
16.   </</span>description> 
17. </</span>property> 

(3)查看方法

在Linux中通过命令行的方式查看
  
文件包括fsimage、edits以及fstime：
fsimage:元数据镜像文件，存储某一时段NameNode内存元数据信息；
edits:操作日志文件；
fstime:保存最近一次checkpoint的时间；
通过浏览器的方式查看
 

1.2.DataNode

(1)作用

提供真实文件数据的存储服务。

(2)Block类以及文件块介绍

通过Ctrl+Shift+T组合键查找“Block”类，必须是位于org.apache.hadoop.hdfs.protocol中

 
文件块（block）：最基本的存储单位。对于文件内容而言，一个文件的长度大小是size，那么从文件的０偏移开始，按照固定的大小，顺序对文件进行划分并编号，划分好的每一个块称一个Block。HDFS默认Block大小是64MB，以一个256MB文件，共有256/64=4个Block.


 
说明：不同于普通文件系统的是，HDFS中，如果一个文件小于一个数据块的大小，并不占用整个数据块存储空间。

(3)Replication(副本)介绍

副本目的在于安全，正是因为集群环境的不可靠，所以才使用副本机制来保证数据的安全性；
副本的缺点在于会占用大量的存储空间，副本越多，占用的空间也会越多。但是相比丢失数据的风险，存储空间的花费是值得的。
副本数一般是三个，从hdfs-default.xml中可以查看：

 

(4)目录结构

从hdfs-default.xml中可以查看该文件在linux上存储位置： 

(5)查看方法

启动服务后在浏览其中输入http://hadoop0:50075/即可访问，参考下图：

 

 

1.3.SecondaryNameNode

作用：合并 NameNode 中的 edits 到 fsimage 中。
执行过程：从NameNode上下载元数据信息（fsimage,edits），然后把二者合并，生成新的fsimage，在本地保存，并将其推送到NameNode，同时重置NameNode的edits.
 
合并原理： 
默认在安装在NameNode节点上，但这样做不安全！

2.Java接口以及常用api

在 hadoop 的 HDFS 操作中，有个非常重要的 api，是 org.apache.hadoop.fs.FileSystem，
这是用户代码操作 HDFS 的直接入口，该类含有操作 HDFS 的各种方法，类似于 jdbc 中操作数据库的直接入口是 Connection 类。

2.1.HDFS之HelloWorld

[java] view plaincopyprint?

1. package mavshuang.helloworld; 
2.  
3. import java.io.IOException; 
4. import java.io.InputStream; 
5. import java.net.URL; 
6.  
7. import org.apache.hadoop.fs.FsUrlStreamHandlerFactory; 
8. import org.apache.hadoop.io.IOUtils; 
9.  
10. public class HelloWorldDemo { 
11.     //设置访问路径 
12.     static final String INPUT_PATH = "hdfs://hadoop0:9000/helloworld"; 
13.  
14.     public static void main(String[] args) throws IOException { 
15.         //设置访问协议 
16.         URL.setURLStreamHandlerFactory(new FsUrlStreamHandlerFactory()); 
17.         final URL url = new URL(INPUT_PATH); 
18.         InputStream inputStream = url.openStream(); 
19.         IOUtils.copyBytes(inputStream, System.out, 1024, true); 
20.     } 
21. } 

2.2.对文件的操作

(1) FileSystem：用户代码操作HDFS时，是直接调用FileSystem的子类完成的

[java] view plaincopyprint?

1. private static FileSystem getFileSystem() throws IOException, URISyntaxException { 
2.     return FileSystem.get(new URI("hdfs://hadoop0:9000/"), new Configuration()); 
3. } 

(2)HDFS的FileSystem之创建文件（夹）

参考代码：

[java] view plaincopyprint?

1. // 创建文件夹 
2. private static void mkDirs(final FileSystem fileSystem) throws IOException { 
3.     boolean mkdirs = fileSystem.mkdirs(new Path("/test")); 
4. } 

在Linux中演示效果：
 

(3)HDFS的FileSystem之上传文件

参考代码：

[java] view plaincopyprint?

1. // 上传文件 
2. private static void uploadFile(final FileSystem fileSystem) throws IOException, FileNotFoundException { 
3.     FSDataOutputStream out = fileSystem.create(new Path(HDFS_PATH)); 
4.     IOUtils.copyBytes(new FileInputStream("C:/Windows/System32/drivers/etc/hosts"), out, new Configuration(), true); 
5. } 

在Linux中演示效果：
 

(4)HDFS的FileSystem之下载文件

参考代码：

[java] view plaincopyprint?

1. // 下载文件 
2. private static void downloadFile(final FileSystem fileSystem) throws IOException { 
3.     FSDataInputStream downloadFile = fileSystem.open(new Path(HDFS_PATH)); 
4.     IOUtils.copyBytes(downloadFile, System.out, new Configuration(), true); 
5. } 

在Eclipse中演示效果：
 
(5)HDFS的FileSystem之查看文件状态
参考代码：

[java] view plaincopyprint?

1. // 查看文件状态 
2. private static void listStatus(final FileSystem fileSystem) throws IOException { 
3.     FileStatus[] listStatus = fileSystem.listStatus(new Path("/")); 
4.     for (FileStatus fileStatus : listStatus) { 
5.         String isDir = fileStatus.isDir() ? "d" : "-"; 
6.         FsPermission permission = fileStatus.getPermission();// 获得文件权限 
7.         short replication = fileStatus.getReplication();// 获得文件副本 
8.         String owner = fileStatus.getOwner();// 获取文件用户 
9.         String group = fileStatus.getGroup();// 获取文件所在组 
10.         long len = fileStatus.getLen();// 获取文件大小 
11.         long modificationTime = fileStatus.getModificationTime();// 获取文件的修改时间 
12.         String path = fileStatus.getPath().toString();// 获取文件所在路径 
13.         String line = isDir + "\t" + permission + "\t" + replication + "\t" + owner + "\t" + group + "\t" + len + "\t" + modificationTime + "\t" + path; 
14.         System.out.println(line); 
15.     } 
16. } 

在Eclipse中演示效果：
 

(6)HDFS的FileSystem之删除文件

参考代码：

[java] view plaincopyprint?

1. //删除文件 
2. private static void delete(final FileSystem fileSystem) throws IOException { 
3.     fileSystem.delete(new Path("/test"),true); 
4. } 

在Linux中演示效果：

 

 

 

 

 

 

 

 

 

# hadoop是适合大数据的分布式存储和计算的平台

# hadoop核心组成由hdfs和mapreduce组成

# hdfs是主从式结构，主节点只有一个，是namenode：从节点有很多个

# 分布式文件系统与HDFS (HDFS体系结构与基本概念)

# Distributed File System

# 数据量越来越多，在一个操作系统管辖的范围存不下了，那么就分配到更多的操作系统管理的磁盘中，但是不方便管理和维护，因此迫切需要一种系统来管理多台机器上的文件，这就是分布式文件管理系统 。

# 是一种允许文件通过网络在多台主机上分享的文件系统，可让多机器上的多用户分享文件和存储空间。

# 通透性。让实际上是通过网络来访问文件的动作，由程序与用户看来，就像是访问本地的磁盘一般。

# 容错。即使系统中有某些节点脱机，整体来说系统仍然可以持续运作而不会有数据损失。

# 分布式文件管理系统很多，hdfs只是其中一种。适用于一次写入多次查询的情况，不支持并发写情况，小文件不合适。

 

# HDFS的shell操作

# 调用文件系统(FS)Shell命令应使用 bin/hadoop fs 的形式。

# 所有的FS shell命令使用URI路径作为参数。

# URI格式是scheme://authority/path。HDFS的scheme是hdfs，对本地文件系统，scheme是file。其中scheme和authority参数都是可选的，如果未加指定，就会使用配置中指定的默认scheme。

# 例如：/parent/child可以表示成hdfs://namenode:namenodePort/parent/child，或者更简单的/parent/child（假设配置文件是namenode:namenodePort）

# 大多数FS Shell命令的行为和对应的Unix Shell命令类似。

 

# HDFS常用命令

# -help [cmd] //显示命令的帮助信息

# -ls(r) <path>      //显示当前目录下所有文件

# -du(s) <path>    //显示目录中所有文件大小

# -count[-q] <path>     //显示目录中文件数量

# -mv <src> <dst> //移动多个文件到目标目录

# -cp <src> <dst>  //复制多个文件到目标目录

# -rm(r)         //删除文件(夹)

# -put <localsrc> <dst>       //本地文件复制到hdfs

# -copyFromLocal //同put

# -moveFromLocal       //从本地文件移动到hdfs

# -get [-ignoreCrc] <src> <localdst>  //复制文件到本地，可以忽略crc校验

# -getmerge <src> <localdst>             //将源目录中的所有文件排序合并到一个文件中

# -cat <src>   //在终端显示文件内容

# -text <src>  //在终端显示文件内容

# -copyToLocal [-ignoreCrc] <src> <localdst>  //复制到本地

# -moveToLocal <src> <localdst>

# -mkdir <path>   //创建文件夹

# -touchz <path>  //创建一个空文件

# HDFS 的shell练习

# hadoop fs -ls /  查看HDFS根目录

# hadoop fs -mkdir /test 在根目录创建一个目录test

# hadoop fs -mkdir /test1 在根目录创建一个目录test1

# echo -e 'hadoop second lesson' >test.txt

# hadoop fs -put ./test.txt /test　或# hadoop fs -copyFromLocal ./test.txt /test

# cd ..

# hadoop fs -get /test/test.txt .   或#hadoopfs -getToLocal /test/test.txt .

# hadoop fs -cp /test/test.txt /test1

# hadoop fs -rm /test1/test.txt

# hadoop fs -mv /test/test.txt /test1

# hadoop fs -rmr /test1  

# NameNode详解

# 是整个文件系统的管理节点。它维护着整个文件系统的文件目录树，文件/目录的元信息和每个文件对应的数据块列表。接收用户的操作请求。

(见源码)

# 文件包括：

# fsimage:元数据镜像文件。存储某一时段NameNode内存元数据信息。

# edits:操作日志文件。

# fstime:保存最近一次checkpoint的时间

# 以上这些文件是保存在linux的文件系统中。

/data/hadoop/hdfs/name

[hadoop@mastername]$ ls

current  image in_use.lock  previous.checkpoint

[hadoop@mastername]$ cd current/

[hadoop@mastercurrent]$ ls

edits  fsimage fstime  VERSION

如果想使用命令hadoopnamenode –format命令格式化namenode发现这个namenode下面的in_use.lock就不能格式化

More in_use.lock 发现是空的 -----表示这个文件存在就是在使用中

说一下current进入这个文件夹

有四个文件有一个VERSION

打开这个

namespace ----命令空间是相对独立的名称空间

namespaceID=-xxxx---- 指的是文件系统的ID

datanode中的块一定要和namenode之间有匹配关系----- 如果两边的namespaceID相同的 --- 这样 通过namespaceID相同 就匹配datanode和

多次执行 hadoop-format 多次之后  出错 因为namespaceID被格式化之后改变了 和datanode中的namespaceID对应不上  ----- 所以第一次之后就出错了

 

# 提示：in_use.lock：如果你想格式化这个namenode的话，hadoop会找这个文件，如果存在的话会提示你Re-format filesystem in /data/hadoop/hdfs/name ? (Y or N) ，以此为凭据知道namenode是否已经格式化过。
# namenode跟secondarynamenode的执行过程

 

# fsimage很重要，所以被备份了

core-default.xml

<property>

  <name>hadoop.tmp.dir</name>

 <value>/tmp/hadoop-${user.name}</value>

  <description>A base for other temporarydirectories.</description>

</property>

hdfs-default.xml

<property>

  <name>dfs.name.dir</name>

 <value>${hadoop.tmp.dir}/dfs/name</value>

  <description>Determines where on thelocal filesystem the DFS name node

      should store the nametable(fsimage).  If this is a comma-delimitedlist

      of directories then the name table isreplicated in all of the

      directories, for redundancy.</description>

</property>

# HA的一个解决方案。但不支持热备。配置即可。

(见源码)

# 执行过程：从NameNode上下载元数据信息（fsimage,edits），然后把二者合并，生成新的# fsimage，在本地保存，并将其推送到NameNode，同时重置NameNode的edits.

默认在安装在NameNode节点上，但这样...不安全！

# Datanode的数据并不是无限存储的，决定与namenode的fsimage，因为fsimage是放在内存中的，内存不可能上百T吧！（内存中放不下了，fsimage设计的时候是被放在内存中的）namenode的元数据信息占内存。

1、 给namenode加内存

2、尽量传大文件。

3、SequenceFile

4、增加block块的大小（这个看环境，就是看一般上传的文件有多大来设计的）

 

 

# java接口及常用api

packagecom.yting.hadoop.hdfs;

 

importjava.io.FileInputStream;

importjava.io.FileNotFoundException;

importjava.io.IOException;

import java.net.URI;

 

importorg.apache.hadoop.conf.Configuration;

importorg.apache.hadoop.fs.FSDataInputStream;

importorg.apache.hadoop.fs.FSDataOutputStream;

importorg.apache.hadoop.fs.FileSystem;

importorg.apache.hadoop.fs.Path;

importorg.apache.hadoop.io.IOUtils;

 

/**

 * 使用FileSystem

 *

 * eclipse创建方法快捷键          Shift+Alt+M 

 * eclipse创建局部变量名称快捷键   Shift+Alt+L

 *

 */

public classHDFSJavaOperation {

       public static final String HDFS_PATH ="hdfs://hadoop0:9000";

       public static final String DIR_PATH ="/d1000";

       public static final String FILE_PATH ="/d1000/f1000";

      

       public static void main(String[] args)throws Exception {

              final FileSystem fileSystem =FileSystem.get(new URI(HDFS_PATH), new Configuration());

             

              //创建文件夹

              //makeDirectory(fileSystem);

             

              //上次文件

              uploadData(fileSystem);

             

              //下载文件 

              //downloadData(fileSystem);

             

              //删除文件(夹)

              //deleteFile(fileSystem);

       }

 

       private static void deleteFile(finalFileSystem fileSystem)

                     throws IOException {

              fileSystem.delete(newPath(FILE_PATH), true);

       }

 

       private static void downloadData(finalFileSystem fileSystem)

                     throws IOException {

              final FSDataInputStream in =fileSystem.open(new Path(FILE_PATH));

              IOUtils.copyBytes(in, System.out,1024, true);

       }

 

       private static void makeDirectory(finalFileSystem fileSystem)

                     throws IOException {

              fileSystem.mkdirs(newPath(DIR_PATH));

       }

 

       private static void uploadData(finalFileSystem fileSystem)

                     throws IOException,FileNotFoundException {

              final FSDataOutputStream out =fileSystem.create(new Path(FILE_PATH));

             

              final FileInputStream in = newFileInputStream("c:/log.txt");

              IOUtils.copyBytes(in, out, 1024,true);

       }

}

# ---------------------------加深拓展----------------------

# RPC调用（后面再做详细分析）

Client发起调用请求，请求调用Server端的对象的方法

 

# MyRpcServer类

package com.yting.hadoop.rpc;

 

importorg.apache.hadoop.conf.Configuration;

import org.apache.hadoop.ipc.RPC;

importorg.apache.hadoop.ipc.RPC.Server;

 

public class MyRpcServer {

    public static String BIND_ADDRESS ="localhost";  // 绑定地址

    public static int PORT = 1129;                   //绑定端口

   

    /** Construct an RPC server. 构造一个RPC的Server

     * @param instance the instance whosemethods will be called实例中的方法被客户端调用的实例

     * @param conf the configuration touse使用的配置

     * @param bindAddress the address tobind on to listen for connection绑定的地址用于监听链接的到来

     * @param port the port to listen forconnections on端口也是用于监听链接的到来

     * @throws Exception

     */

    public static void main(String[] args)throws Exception {

       MyInstance myInstance = new MyInstanceImpl();

       final Server server = RPC.getServer(myInstance,BIND_ADDRESS,PORT,new Configuration());

       server.start();

    }

}

# MyRpcClient类

package com.yting.hadoop.rpc;

 

import java.net.InetSocketAddress;

 

importorg.apache.hadoop.conf.Configuration;

import org.apache.hadoop.ipc.RPC;

 

public class MyRpcClient {

    public static void main(String[] args)throws Exception {

       /** Construct a client-side proxy object that implements the named protocol,

          * talking to a server at the named address.*/

       /*

        * Class<? extends VersionedProtocol> protocol,

         * long clientVersion,

         * InetSocketAddress addr,

         * Configuration conf

        */

       MyInstance proxy =(MyInstance) RPC.waitForProxy(MyInstance.class, MyInstance.versionID,newInetSocketAddress(MyRpcServer.BIND_ADDRESS, MyRpcServer.PORT),newConfiguration());

       String retVal = proxy.hello("world");

       System.out.println("客户端调用结果：" + retVal);

       RPC.stopProxy(proxy);

    }

}

# MyInstance接口

package com.yting.hadoop.rpc;

 

import org.apache.hadoop.ipc.VersionedProtocol;

 

public interface MyInstance extends VersionedProtocol{

    public static final long versionID = 1234567L;

   

    public abstract String hello(String name);

 

}

# MyInstanceImpl实现

package com.yting.hadoop.rpc;

 

import java.io.IOException;

 

public class MyInstanceImpl  implements MyInstance{

    /* (non-Javadoc)

     * @seecom.yting.hadoop.rpc.MyInstance#hello(java.lang.String)

     */

    @Override

    public String hello(String name) {

       System.out.println("我被调用了、、、");

       return"hello" + name;

    }

 

    @Override

    public long getProtocolVersion(String protocol, long clientVersion)throws IOException {

       return MyInstance.versionID;

    }

}

# 运行结果

客户端的运行结果

客户端调用结果：helloworld

服务端的运行结果

14/03/02 15:35:42 INFO ipc.Server: Starting SocketReader

14/03/02 15:35:42 INFO ipc.Server: IPC Server Responder: starting

14/03/02 15:35:42 INFO ipc.Server: IPC Server listener on 1129: starting

14/03/02 15:35:42 INFO ipc.Server: IPC Server handler 0 on 1129: starting

我被调用了、、、

# 结论

1、 RPC 实际上就是RPC远程过程调用

2、 被调用的对象位于服务端，并且这个对象必须有接口（jdk反射要求），实现VersionedProtocol（api要求）

3、 客户端调用的对象中的方法必须位于接口中

4、在本地运行jps看看

 

由此可以推断出hadoop中启动的5个进程，也就是RPC的服务端

# HDFS的分布式存储架构的分析（后面分析FileSystem的源代码）

# HDFS的高可靠

Fsimage备份

Secondarynamenode

# edits文件可以没有么？（必须有）下面是一个例子

F1 starttransfer

F1 block1 shuff

F1 end transfer

Edits文件仅仅记录操作日志(确保事务的正确性)

 

 

 

 

 

 

 

 

 

 

 

 

 

 

 

 

 

 

 

 

 

 

 

 

 

 

# Day3 介绍MapReduce体系结构及各种算法(1)

# MapReduce的介绍

# MapReduce是Hadoop的分布式计算框架，由两个阶段组成，分别是map和reduce阶段，对于程序员而言，使用过程非常简单，只要覆盖map阶段中的map方法和reduce节点的reduce方法即可

# map和reduce阶段的形参的键值对的形式

# mapreduce的执行流程

 

瓶颈：磁盘IO

# mapreduce执行原理

 

1.1 读取输入文件内容，解析成key、value对。对输入文件的每一行，解析成key、value对。每一个键值对调用一次map函数。

1.2 写自己的逻辑，对输入的key、value处理，转换成新的key、value输出。

1.3 对输出的key、value进行分区。

1.4 对不同分区的数据，按照key进行排序、分组。相同key的value放到一个集合中。

1.5 (可选)分组后的数据进行归约。(Combine)

2.0 reduce任务处理

2.1 对多个map任务的输出，按照不同的分区，通过网络copy到不同的reduce节点。

2.2 对多个map任务的输出进行合并、排序。写reduce函数自己的逻辑，对输入的key、value处理，转换成新的key、value输出。

2.3 把reduce的输出保存到文件中。

例子：实现WordCountApp

# 第一个统计单词的java程序（hadoop自带的例子源码）

packageorg.apache.hadoop.examples;

 

importjava.io.IOException;

importjava.util.StringTokenizer;

 

importorg.apache.hadoop.conf.Configuration;

importorg.apache.hadoop.fs.Path;

importorg.apache.hadoop.io.IntWritable;

importorg.apache.hadoop.io.Text;

importorg.apache.hadoop.mapreduce.Job;

importorg.apache.hadoop.mapreduce.Mapper;

importorg.apache.hadoop.mapreduce.Reducer;

importorg.apache.hadoop.mapreduce.lib.input.FileInputFormat;

importorg.apache.hadoop.mapreduce.lib.output.FileOutputFormat;

importorg.apache.hadoop.util.GenericOptionsParser;

 

@SuppressWarnings("all")

public classWordCount {

 

       public static class TokenizerMapperextends Mapper<Object, Text, Text, IntWritable> {

 

              private final static IntWritableone = new IntWritable(1);

              private Text word = new Text();

 

              public void map(Object key, Textvalue, Context context) throws IOException, InterruptedException {

                     StringTokenizer itr = newStringTokenizer(value.toString());

                     while (itr.hasMoreTokens()){

                            word.set(itr.nextToken());

                            context.write(word,one);

                     }

              }

       }

 

       public static class IntSumReducer extendsReducer<Text, IntWritable, Text, IntWritable> {

              private IntWritable result = newIntWritable();

 

              public void reduce(Text key,Iterable<IntWritable> values,

                            Context context)throws IOException, InterruptedException {

                     int sum = 0;

                     for (IntWritable val :values) {

                            sum += val.get();

                     }

                     result.set(sum);

                     context.write(key, result);

              }

       }

 

       public static void main(String[] args)throws Exception {

              Configuration conf = newConfiguration();

              String[] otherArgs = newGenericOptionsParser(conf, args).getRemainingArgs();

              if (otherArgs.length != 2) {

                     System.err.println("Usage:wordcount <in> <out>");

                     System.exit(2);

              }

              Job job = new Job(conf, "wordcount");

              job.setJarByClass(WordCount.class);

              job.setMapperClass(TokenizerMapper.class);

              job.setCombinerClass(IntSumReducer.class);

              job.setReducerClass(IntSumReducer.class);

              job.setOutputKeyClass(Text.class);

              job.setOutputValueClass(IntWritable.class);

              FileInputFormat.addInputPath(job,new Path(otherArgs[0]));

              FileOutputFormat.setOutputPath(job,new Path(otherArgs[1]));

              System.exit(job.waitForCompletion(true)? 0 : 1);

       }

}

# 下面运行命令跟输出结果

[hadoop@masterhadoop-1.1.2]$ hadoop jar hadoop-yting-wordcounter.jarorg.apache.hadoop.examples.WordCount /user/hadoop/20140303/test.txt/user/hadoop/20140303/output001

14/03/0310:43:51 INFO input.FileInputFormat: Total input paths to process : 1

14/03/0310:43:52 INFO mapred.JobClient: Running job: job_201403020905_0001

14/03/0310:43:53 INFO mapred.JobClient:  map 0%reduce 0%

14/03/0310:44:12 INFO mapred.JobClient:  map 100%reduce 0%

14/03/03 10:44:25INFO mapred.JobClient:  map 100% reduce100%

14/03/0310:44:29 INFO mapred.JobClient: Job complete: job_201403020905_0001

14/03/0310:44:29 INFO mapred.JobClient: Counters: 29

14/03/0310:44:29 INFO mapred.JobClient:   JobCounters

14/03/03 10:44:29INFO mapred.JobClient:     Launchedreduce tasks=1

14/03/0310:44:29 INFO mapred.JobClient:    SLOTS_MILLIS_MAPS=19773

14/03/0310:44:29 INFO mapred.JobClient:     Totaltime spent by all reduces waiting after reserving slots (ms)=0

14/03/0310:44:29 INFO mapred.JobClient:     Totaltime spent by all maps waiting after reserving slots (ms)=0

14/03/0310:44:29 INFO mapred.JobClient:    Launched map tasks=1

14/03/0310:44:29 INFO mapred.JobClient:    Data-local map tasks=1

14/03/0310:44:29 INFO mapred.JobClient:    SLOTS_MILLIS_REDUCES=13148

14/03/0310:44:29 INFO mapred.JobClient:   FileOutput Format Counters

14/03/0310:44:29 INFO mapred.JobClient:     BytesWritten=188

14/03/0310:44:29 INFO mapred.JobClient:  FileSystemCounters

14/03/0310:44:29 INFO mapred.JobClient:    FILE_BYTES_READ=171

14/03/0310:44:29 INFO mapred.JobClient:    HDFS_BYTES_READ=310

14/03/0310:44:29 INFO mapred.JobClient:    FILE_BYTES_WRITTEN=101391

14/03/0310:44:29 INFO mapred.JobClient:    HDFS_BYTES_WRITTEN=188

14/03/0310:44:29 INFO mapred.JobClient:   FileInput Format Counters

14/03/0310:44:29 INFO mapred.JobClient:     BytesRead=197

14/03/0310:44:29 INFO mapred.JobClient:  Map-Reduce Framework

14/03/0310:44:29 INFO mapred.JobClient:     Mapoutput materialized bytes=163

14/03/0310:44:29 INFO mapred.JobClient:     Mapinput records=8

14/03/0310:44:29 INFO mapred.JobClient:    Reduce shuffle bytes=163

14/03/0310:44:29 INFO mapred.JobClient:    Spilled Records=56

14/03/0310:44:29 INFO mapred.JobClient:     Map output bytes=376

14/03/0310:44:29 INFO mapred.JobClient:     CPUtime spent (ms)=4940

14/03/0310:44:29 INFO mapred.JobClient:     Totalcommitted heap usage (bytes)=63926272

14/03/0310:44:29 INFO mapred.JobClient:    Combine input records=45

14/03/0310:44:29 INFO mapred.JobClient:    SPLIT_RAW_BYTES=113

14/03/0310:44:29 INFO mapred.JobClient:    Reduce input records=28

14/03/0310:44:29 INFO mapred.JobClient:    Reduce input groups=28

14/03/0310:44:29 INFO mapred.JobClient:    Combine output records=28

14/03/0310:44:29 INFO mapred.JobClient:    Physical memory (bytes) snapshot=111722496

14/03/0310:44:29 INFO mapred.JobClient:    Reduce output records=28

14/03/0310:44:29 INFO mapred.JobClient:    Virtual memory (bytes) snapshot=468000768

14/03/0310:44:29 INFO mapred.JobClient:     Mapoutput records=45

[hadoop@masterhadoop-1.1.2]$ hadoop fs -ls /user/hadoop/20140303/output001

Found 3 items

-rw-r--r--   1 hadoop supergroup          0 2014-03-03 10:44/user/hadoop/20140303/output001/_SUCCESS

drwxr-xr-x   - hadoop supergroup          0 2014-03-03 10:43/user/hadoop/20140303/output001/_logs

-rw-r--r--   1 hadoop supergroup        188 2014-03-03 10:44/user/hadoop/20140303/output001/part-r-00000

[hadoop@masterhadoop-1.1.2]$ hadoop fs -text /user/hadoop/20140303/output001/part-t-00000

text: File doesnot exist: /user/hadoop/20140303/output001/part-t-00000

[hadoop@masterhadoop-1.1.2]$ hadoop fs -text /user/hadoop/20140303/output001/part-r-00000

a      1

again       1

and  1

changce   1

easy 1

forever    1

give 1

hand       1

heart       2

hold 1

i      1

is     1

it     1

love 1

me   6

meimei    1

miss 1

see   1

show       1

smile       1

so    1

soul 1

take 3

the   2

to    4

until 1

what       1

you  6

# 最小的MapReduce（默认设置）

Configurationconfiguration = new Configuration();

Job job = newJob(configuration, "HelloWorld");

job.setInputFormat(TextInputFormat.class);

job.setMapperClass(IdentityMapper.class);

job.setMapOutputKeyClass(LongWritable.class);

job.setMapOutputValueClass(Text.class);

job.setPartitionerClass(HashPartitioner.class);

job.setNumReduceTasks(1);

job.setReducerClass(IdentityReducer.class);

job.setOutputKeyClass(LongWritable.class);

job.setOutputValueClass(Text.class);

job.setOutputFormat(TextOutputFormat.class);

job.waitForCompletion(true);

 

# 序列化

# Writable

# 数据流单向的

# LongWritable不能进行加减等操作（没必要，java的基本类型都已经弄了这些功能了）

# JobTracker，TaskTracker

# JobTracker

负责接收用户提交的作业，负责启动、跟踪任务执行。

JobSubmissionProtocol是JobClient与JobTracker通信的接口。

InterTrackerProtocol是TaskTracker与JobTracker通信的接口。

# TaskTracker

负责执行任务

# JobClient

是用户作业与JobTracker交互的主要接口。

负责提交作业的，负责启动、跟踪任务执行、访问任务状态和日志等。

# 执行过程

图片2.2M就上传不了了，真是无语、、、

 

 

 

 

 

 

http://blog.csdn.net/u012185296/article/details/20491381

http://blog.sina.com.cn/s/blog_65309a0f0102v9m7.html