java对HDFS中文件的操作

 

 

1 概述：

1   1     Hadoop 提供一类api可以通过java工程操作hdfs中的文件，包括：文件打开，读写，删除等、

           目录的创建，删除，读取目录中所有文件等。

2   2    需要处理hadoop的文件程序需要引入下面jar包，可以hadoop-1.2.1\lib找到。

     hadoop-core和common-log，commons-configuration-1.6.jar，commons-lang-2.4.jar。

 

3  3 想要运行写好的程序操作hdfs中文件两种方法：1 安装elipse的hadoop插件（之前有介绍）  2、将写好成程序打成jar包，在服务器上用hadoop命令运行

 

4  4  程序处理步骤： 1） 得到Configuration对象2）得到FileSystem对象 3）进行文件操作。

 

    5 以下是操作文件的事例程序，版本 hadoop-1.2.1

 

package org.tony.hdfs;

import java.io.IOException;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FSDataInputStream;
import org.apache.hadoop.fs.FSDataOutputStream;
import org.apache.hadoop.fs.FileStatus;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;


public class HDFSTest {
	public static void main(String[] args) throws Exception
	{
		//uploadLocalFile2HDFS("E:/1.txt","/tmp/1.txt");//E盘下文件传到hdfs上
		//createNewHDFSFile("/tmp/create2", "hello");
		//String str = new String(readHDFSFile("/tmp/create2"));
		//System.out.println(str);
		
		//mkdir("/tmp/testdir");
		//deleteDir("/tmp/testdir");
		//listAll("/tmp/");
                getDateNodeHost();
	}
	//获取HDFS集群上所有节点名称信息
	public static void getDateNodeHost() throws IOException{
		
		Configuration conf = getConf();

               FileSystem fs=FileSystem.get(conf);
               DistributedFileSystem hdfs = (DistributedFileSystem)fs;
               DatanodeInfo[] dataNodeStats = hdfs.getDataNodeStats();
               for(int i=0;i<dataNodeStats.length;i++){
                  System.out.println("DataNode_"+i+"_Name:"+dataNodeStats[i].getHostName());
               }
	}
	/*
	 * upload the local file to the hds 
	 * 路径是全路径
	 */
	public static void uploadLocalFile2HDFS(String s, String d) 
		throws IOException
	{
		Configuration conf = getConf();
		FileSystem hdfs = FileSystem.get(conf);
		Path src = new Path(s);
		Path dst = new Path(d);
		hdfs.copyFromLocalFile(src, dst);
		hdfs.close();
	}
	
	/*
	 * create a new file in the hdfs.
	 * notice that the toCreateFilePath is the full path
	 * and write the content to the hdfs file.
	 */
	public static void createNewHDFSFile(String toCreateFilePath, String content) throws IOException
	{
		Configuration conf = getConf();
		FileSystem hdfs = FileSystem.get(conf);
		
		FSDataOutputStream os = hdfs.create(new Path(toCreateFilePath));
		os.write(content.getBytes("UTF-8"));
		os.close();
		hdfs.close();
	}
	
	/*
	 * delete the hdfs file 
	 * notice that the dst is the full path name
	 */
	public static boolean deleteHDFSFile(String dst) throws IOException
	{
		Configuration conf = getConf();
		FileSystem hdfs = FileSystem.get(conf);
		
		Path path = new Path(dst);
		boolean isDeleted = hdfs.delete(path);
		hdfs.close();
		return isDeleted;
	}
	
	
	/*
	 * read the hdfs file content
	 * notice that the dst is the full path name
	 */
	public static byte[] readHDFSFile(String dst) throws Exception
	{
		Configuration conf = getConf();
		FileSystem fs = FileSystem.get(conf);
		
		// check if the file exists
		Path path = new Path(dst);
		if ( fs.exists(path) )
		{
			FSDataInputStream is = fs.open(path);
			// get the file info to create the buffer
			FileStatus stat = fs.getFileStatus(path);
			// create the buffer
			byte[] buffer = new byte[Integer.parseInt(String.valueOf(stat.getLen()))];
		    is.readFully(0, buffer);
		    
		    is.close();
		    fs.close();
		    
		    return buffer;
		}
		else
		{
			throw new Exception("the file is not found .");
		}
	}
	
	
	/*
	 * make a new dir in the hdfs
	 * the dir may like '/tmp/testdir'
	 */
	public static void mkdir(String dir) throws IOException
	{
		Configuration conf = getConf();
		FileSystem fs = FileSystem.get(conf);
		fs.mkdirs(new Path(dir));
		
		fs.close();
	}
	
	/*
	 * delete a dir in the hdfs
	 * dir may like '/tmp/testdir'
	 */
	public static void deleteDir(String dir) throws IOException
	{
		Configuration conf = getConf();
		FileSystem fs = FileSystem.get(conf);
		fs.delete(new Path(dir));
		fs.close();
	}
	//文件系统连接到 hdfs的配置信息 
	private static Configuration getConf(){
		Configuration conf = new Configuration();
		// 这句话很关键，这些信息就是hadoop配置文件中的信息
		conf.set("mapred.job.tracker", "192.168.102.136:9001");
		conf.set("fs.default.name", "hdfs://192.168.102.136:9000");
		return conf;
	}
	
	/**
	* @Title: listAll 
	* @Description: 列出目录下所有文件 
	* @return void    返回类型 
	* @throws
	 */
	public static void listAll(String dir) throws IOException
	{
		Configuration conf = getConf();
		FileSystem fs = FileSystem.get(conf);
		FileStatus[] stats = fs.listStatus(new Path(dir));
		for(int i = 0; i < stats.length; ++i)
		{
			if (!stats[i].isDir())
			{
				// regular file
				System.out.println(stats[i].getPath().toString());
			}
			else 
			{
				// dir
				System.out.println(stats[i].getPath().toString());
			}
//			else if(stats[i].())
//			{
//				// is s symlink in linux
//				System.out.println(stats[i].getPath().toString());
//			}
 				
		}
		fs.close();
	}
	
}

 

 注意：因为hadoop读写文件时候需要权限，如果你window用户名为admin而hadoop集群中目录权限是hadoop用户，读写会失败

  1 修改window系统用户为hadoop

  2 elipse运行的jvm参数中设置 -DHADOOP_USER_NAME=hadoop ，运行的用户为hadoop 

  3 部署一个linux环境，在linux环境中开发是比较容易的（用户问题容易设置）。

  4 运行的时候指定用户 ，如下：

   Configuration conf = new Configuration();

FileSystem hdfs = FileSystem.get(new URI("hdfs://192.168.142.133:9000"),conf,"hadoop");

 

  

 

 

 

评论

1 楼 胖好汉 2015-09-03  

不知道你运行过程序没有，会报错啊。。。