HadoopOutline的示例分析

小编给大家分享一下Hadoop Outline的示例分析，希望大家阅读完这篇文章之后都有所收获，下面让我们一起去探讨吧！

成都创新互联凭借在网站建设、网站推广领域领先的技术能力和多年的行业经验，为客户提供超值的营销型网站建设服务，我们始终认为：好的营销型网站就是好的业务员。我们已成功为企业单位、个人等客户提供了成都网站制作、网站设计服务，以良好的商业信誉，完善的服务及深厚的技术力量处于同行领先地位。

Hdfs Java API Sample

Read by hadoop FsURLStreamHandlerFactory
Read/Write by hadoop DistributeFileSystem

Hadoop Outline的示例分析

package com.jinbao.hadoop.hdfs;

import java.io.IOException;
import java.io.InputStream;
import java.io.OutputStream;
import java.net.URI;
import java.net.URL;

import org.apache.commons.io.IOUtils;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FSDataOutputStream;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.FsUrlStreamHandlerFactory;
import org.apache.hadoop.fs.Path;

/**
 * 
 */

/**
 * @author cloudera
 *
 */
public class HdfsClient {
	static String sFileUrl = "hdfs://quickstart.cloudera/gis/gistool/README.md";

	/**
	 * @param args
	 * @throws IOException 
	 */
	public static void main(String[] args) throws IOException {
		if(args.length >= 2){
			String sUrl = sFileUrl;
			if(args[0].equalsIgnoreCase("-r-url")){
				sUrl = args[1];
				
				//test read by hadoop FsURLStreamHandlerFactory 
				readHdfsFileByDfsUrl(sUrl);
			}
			else if(args[0].equalsIgnoreCase("-r-file")){
				sUrl = args[1];
				//test read by hadoop dfsFile
				readHdfsFileByDfsFileApi(sUrl);
			}
			else if(args[0].equalsIgnoreCase("-w-file")){
				sUrl = args[1];
				//test read by hadoop dfsFile
				writeHdfsFileByDfsFileApi(sUrl);
			}
			else if(args[0].equalsIgnoreCase("-w-del")){
				sUrl = args[1];
				//test read by hadoop dfsFile
				deleteHdfsFileByDfsFileApi(sUrl);
			}
		}
		

		
	}

	private static void deleteHdfsFileByDfsFileApi(String sUrl) {
		Configuration conf = new Configuration();

		try {
			
			FileSystem fs = FileSystem.get(URI.create(sUrl),conf);
			Path path = new Path(sUrl);
			fs.delete(path,true);
			
		} catch (IOException e) {
			e.printStackTrace();
		}
		finally{
		}
	}

	private static void writeHdfsFileByDfsFileApi(String sUrl) {
		
		Configuration conf = new Configuration();
		OutputStream out = null;
		byte[] data = "Writing Test".getBytes(); 
		// Get a FSDataInputStream object
		try {
			// Get a FSDataInputStream object, actually is HdfsDataInputSteam
			FileSystem fs = FileSystem.get(URI.create(sUrl),conf);
			Path path = new Path(sUrl);
			if(fs.exists(path)){
				out = fs.append(path);
				IOUtils.write(data, out);

			}
			else{
				out = fs.create(path);

				out.write(data);
				// flush buffer to OS
				out.flush();
				FSDataOutputStream fsout = FSDataOutputStream.class.cast(out);
				// Sync data to disk
				fsout.hsync();
				
				// call sync implicitly
				out.close();
			}
		} catch (IOException e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		}
		finally{
			IOUtils.closeQuietly(out);
		}
		
	}

	public static void readHdfsFileByDfsUrl(String sUrl){
		
		URL.setURLStreamHandlerFactory(new FsUrlStreamHandlerFactory());
		
		InputStream in = null;
		try{
			URL url = new URL(sUrl);
			in = url.openStream();
			IOUtils.copy(in,System.out);
		}
		catch(IOException ioe){
			ioe.printStackTrace();
		}
		finally{
			IOUtils.closeQuietly(in);
		}
	}
	
	private static void readHdfsFileByDfsFileApi(String sUrl) {
		
		Configuration conf = new Configuration();
		
		InputStream in = null;
		
		try{
			FileSystem fs = FileSystem.get(URI.create(sUrl),conf);

			//Get a FSDataInputStream object, actually HdfsDataInputStream			
			in = fs.open(new Path(sUrl));
			IOUtils.copy(in,System.out);
		}
		catch(IOException ioe){
			ioe.printStackTrace();
		}
		finally{
			IOUtils.closeQuietly(in);
		}	
	}

}

Data Flow on Read

1. FileSystem to get DistributeFileSystem

2. DistributeFileSystem通过和namenode调用，获得前面几个块位置，并返回datanode的地址，包括备用节点。

3. 客户端调用DistributeFileSystem获取FSDataInputStream,这个流对象连接距离最近的datanode,通过调用Read读取数据。

4. 如果当前块已经读完，则FSDataInputStream关闭这个块，并寻找下一个最佳的datanode并继续读取。

5. 读完最后一个块后，调用close方法关闭数据流。

容错处理

1. FSDataInputStream如果与datanode通信遇到错误，则尝试下一个最佳备用节点。另外，也会记住那个坏节点，以后不会在它上面读数据。并且，会通知namenode哪一个节点有问题。

Data Flow on Write

1. FileSystem to get DistributeFileSystem

2. DistributeFileSystem通过和namenode的RPC调用create()方法，创建文件原数据。如果已经存在，报出IOException.

3. 客户端调用DistributeFileSystem获取FSDataOutputStream,它封装了一个DFSOutputStream对象，来处理datanode和namenode通信。

4. 客户端开始写数据，则FSDataOutputStream将当前块数据分成一个个的数据包packet,并写入一个数据包队列(Data packet queue), 然后datastreamer来根据datanode列表，要求namenode分配合适的datanode，DataStreamer把这些datanode组成数据管线 (datanode pipeline),数目有dfs.replication决定。

5. 开始写数据，每写入一个包都将它备份到另一个确认队列(Data Ack queue)，第一个被写入的节点，会把数据写入第二个节点，然后第三个。如果收到都写完的通知，则从确认队列中删除。

6. 写完一个块后，重复4-5，直到最后写完，调用close方法关闭数据流。

容错处理

FSDataOutputStream如果与datanode通信遇到错误

1. 关闭pipeline

2. 把确认队列的数据包，添加到数据包队列，以防止下游节点(downstream node)丢失数据。

3. 为存储正常的datanode的当前数据制定一个新的标示(identifier), 并把这个标示传递给namenode，以便namenode删除故障node的部分数据。

4. 把剩余的数据写入剩下的好的datanode。namenode会创建新的节点，来复制数据，以达到复本量。对于当前写过程，如果写入成功的节点达到dfs.replication.min就算成功，其他的由namenode进行复制。

复本的布局

照顾稳定性和负载均衡

Hadoop的默认布局策略是在运行客户端上放置第1个复本,如果客户端在cluster外,则在集群中随机选择一个节点.

第2个和第3个会随机选择另外一个相同Rack上的两个节点.

分布式复制 distcp

% hadoop distcp hdfs://namenode/foo hdfs://namenode2/foo

distcp使用map-reduce的作业来实现，非常适用于两个数据中心同步数据。

如果两个数据中心版本不一致，可以试用hftp协议，使得作业之运行在目标系统上

% hadoop distcp hftp://namenode:50070/foo hdfs://namenode2/foo

注:需要指定hftp端口 50070

为了使map平衡集群，可以参考N*20的设置：-m 20*N,N是节点总数.

balancer还没看。

归档工具Har

减小namenode的内存，适合管理小文件，它还是透明的，对map-reduce也是适用的。

%hadoop archive -archivename files.har /myfiles/ /my

不足:

har相当于tar功能，可以打包文件，不支持压缩。仅仅节省的namenode的内存。

一旦创建就不能修改，想添加和删除文件，必须重新建立har.

看完了这篇文章，相信你对“Hadoop Outline的示例分析”有了一定的了解，如果想了解更多相关知识，欢迎关注创新互联行业资讯频道，感谢各位的阅读！