Hadoop数据流原理+实例代码

从HDFS中读取文件test.txt

前提：启动Hadoop的所有组件

⑴准备test.txt

查询tetst.txt文件是否存在：hadoop fs -ls hdfs://localhost/test/

注意：本人的test.txt文件在/test目录下，根据自己的实际查找，如果没有则创建一个。如下是test.txt文件中的内容

[ld@localhost /]$ hadoop fs -cat hdfs://localhost/test/test.txt
hello world 
===========
     -- name : liudong

⑵编写代码（在windows下的eclipise中完成，注意jar包的导入）

package com.dong.hello;

import java.io.BufferedInputStream;
import java.io.FileInputStream;
import java.io.IOException;
import java.io.InputStream;
import java.io.OutputStream;
import java.net.URI;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.IOUtils;

public class FileSystemCat {
	
	public static void main(String[] args) throws IOException{
		//从HDFS上读取test.txt文件，显示到终端
		String uri="hdfs://localhost/test/test.txt";
		Configuration conf = new Configuration();
		FileSystem fs = FileSystem.get(URI.create(uri), conf);
		InputStream in = null;
                try{
		in = fs.open(new Path(uri));
		IOUtils.copyBytes(in, System.out, 4096,false);
                }finally{
		IOUtils.closeStream(in);
                }
	
	}
	
		
}

⑶导出成jar包的形式（这里一定要注意windows中JDK的版本和Hadoop中的JDK的版本一致，不然不能运行）

⑷上传到hadoop所在的主机（这里我的是Centos)(使用软件为WinSCP)

⑸在Centos中使用hadoop命令进行执行

[ld@localhost ~]$ hadoop jar hadoopHello.jar 
hello world 
===========
     -- name : liudong

（如果现实没有找到主类，则注意HADOOP_CLASSPATH的配置，或者jar包）

Hadoop数据流原理：

客户端通过调用FileSystem对象的open()方法打开目标文件，从HDFS的角度出发：

DistributedFileSystem通过使用RFC调用namenode,来确定文件起始块的位置，namenode返回存有该该块副本的datanode地址，如果客户端本身就是一个datanode，那客户端会从保存有相应数据块复本的本地datanode读取数据。DistributedFileSystem类返回一个FSDataInputStream对象让客户端以便读取数据，FSDataInputStream类封装DFSInputStream对象，DSFInputStream对象管理着datanode,namenode的I/O.

然后客户端对输入流调用read()方法，存储着文件起始块的datanode地址的DFSInputStream随即连接距离最近的文件中第一个块所在的datanode.通过对数据流反复调用read()方法，可以将数据从datanode传输到客户端，当到达块的末端时，DFSInputStream关闭与该datanode的连接，寻找下一个块的最佳的datanode，当客户端完成读取，就对FSDataInputStream调用close()方法。

在读取的过程，如果DFSInputStream在于datanode出错，会尝试从这个块最近的datanode读取数据，同时记住这个故障datanode,保证以后不会反复读取该节点上后续的块。

客户端通过调用create()方法来创建文件，从HDFS的角度出发：

DistributedFileSystem对namenode创建一个RFC调用，在文件系统的命名空间中新建一个文件，此时该文件中还没有相应的数据块，namenode执行各种不同的检查来确保这个文件不存在以及客户端是否创建文件的权限，如果这些检查通过，namenode就会为创建新文件记录一条记录，否则创建失败，并像客户端抛出一个IOException异常。

DistributedFileSystem向客户端返回一个FSDataOutputStream对象，由此客户端可以开始写入数据，在客户端写入数据时，DFSOutputStream将它分成一个个数据包，并且写入内部队列，称为数据队列，DataStreamer处理数据队列，它负责挑选出合适存储数据复本的一组datanode，并以此来要求namenode分配新的数据块，这一组datanode构成一个管线，假设复本数为3，所以管线中有3个节点，DataStreamer将数据包流式传输到管线中第一个datanode，该datanode存储数据包并将它发送到管线中第二个datanode，同样，第二个datanode存储数据包并且发送到管线中第三个datanode。

同使用DFSOutputStream维护一个内部数据包队列来等待datanode的收到确认回执，称为“确认队列”，收到管道中所有datanode确认信息后，该数据包才会从确认队列删除。如果datanode在数据写入的时候发生故障，则先关闭管线，确认把队列中所有数据包都添加回数据队列的最前端，以确保故障节点下游的datanode不会漏掉任何一个数据包。同时将正常datanode的标识传送给namenode，以便故障datanode在恢复后可以删除存储在部分数据块。从管线中删除故障datanode，基于两个正常datanode构建一个新的管线，剩下的数据块写入管线中正常的datanode,namenode注意到块副本量不足时，会在另一个节点上创建新的复本。客户端完成数据的写入后，对数据流调用close()方法。