HDFS

weitao1026

浏览: 995072 次
性别:
来自: 上海

最近访客更多访客>>

vicento4

723499280

liuzidong

s1986q

博主相关

博客

微博

相册

留言

关于我

文章分类

社区版块

存档分类

博客分类：

hadoop/spark/Hbase/Hive/pig/Zookeeper/HAWQ/cloudera/Impala/Oozie

hadoop

HDFS是Hadoop生态系统的根基，也是Hadoop生态系统中的重要一员，大部分时候，我们都会使用Linux shell命令来管理HDFS，包括一些文件的创建，删除，修改，上传等等，因为使用shell命令操作HDFS的方式，相对比较简单，方便，但是有时候，我们也需要通过编程的方式来实现对文件系统的管理。

比如有如下的一个小需求，要求我们实现读取HDFS某个文件夹下所有日志，经过加工处理后在写入到HDFS上，或者存进Hbase里，或者存进其他一些存储系统。这时候使用shell的方式就有点麻烦了，所以这时候我们就可以使用编程的方式来完成这件事了，当然散仙在这里使用的是原生的Java语言的方式，其他的一些语言例如C++,PHP,Python都可以实现，散仙在这里不给出演示了，（其实散仙也不会那些语言，除了刚入门的Python）。

下面，散仙给出代码，以供参考：

Java代码

package com.java.api.hdfs;
import java.io.BufferedReader;
import java.io.IOException;
import java.io.InputStream;
import java.io.InputStreamReader;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileStatus;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
/**
* @author 三劫散仙
* Java API操作HDFS
* 工具类
*
* **/
public class OperaHDFS {
public static void main(String[] args)throws Exception {
//System.out.println("aaa");
// uploadFile();
//createFileOnHDFS();
//deleteFileOnHDFS();
//createDirectoryOnHDFS();
//deleteDirectoryOnHDFS();
// renameFileOrDirectoryOnHDFS();
//downloadFileorDirectoryOnHDFS();
readHDFSListAll();
}
/***
* 加载配置文件
* **/
static Configuration conf=new Configuration();
/**
* 重名名一个文件夹或者文件
*
* **/
public static void renameFileOrDirectoryOnHDFS()throws Exception{
FileSystem fs=FileSystem.get(conf);
Path p1 =new Path("hdfs://10.2.143.5:9090/root/myfile/my.txt");
Path p2 =new Path("hdfs://10.2.143.5:9090/root/myfile/my2.txt");
fs.rename(p1, p2);
fs.close();//释放资源
System.out.println("重命名文件夹或文件成功.....");
}
/***
*
* 读取HDFS某个文件夹的所有
* 文件，并打印
*
* **/
public static void readHDFSListAll() throws Exception{
//流读入和写入
InputStream in=null;
//获取HDFS的conf
//读取HDFS上的文件系统
FileSystem hdfs=FileSystem.get(conf);
//使用缓冲流，进行按行读取的功能
BufferedReader buff=null;
//获取日志文件的根目录
Path listf =new Path("hdfs://10.2.143.5:9090/root/myfile/");
//获取根目录下的所有2级子文件目录
FileStatus stats[]=hdfs.listStatus(listf);
//自定义j，方便查看插入信息
int j=0;
for(int i = 0; i < stats.length; i++){
//获取子目录下的文件路径
FileStatus temp[]=hdfs.listStatus(new Path(stats[i].getPath().toString()));
for(int k = 0; k < temp.length;k++){
System.out.println("文件路径名:"+temp[k].getPath().toString());
//获取Path
Path p=new Path(temp[k].getPath().toString());
//打开文件流
in=hdfs.open(p);
//BufferedReader包装一个流
buff=new BufferedReader(new InputStreamReader(in));
String str=null;
while((str=buff.readLine())!=null){
System.out.println(str);
}
buff.close();
in.close();
}
}
hdfs.close();
}
/**
* 从HDFS上下载文件或文件夹到本地
*
* **/
public static void downloadFileorDirectoryOnHDFS()throws Exception{
FileSystem fs=FileSystem.get(conf);
Path p1 =new Path("hdfs://10.2.143.5:9090/root/myfile//my2.txt");
Path p2 =new Path("D://7");
fs.copyToLocalFile(p1, p2);
fs.close();//释放资源
System.out.println("下载文件夹或文件成功.....");
}
/**
* 在HDFS上创建一个文件夹
*
* **/
public static void createDirectoryOnHDFS()throws Exception{
FileSystem fs=FileSystem.get(conf);
Path p =new Path("hdfs://10.2.143.5:9090/root/myfile");
fs.mkdirs(p);
fs.close();//释放资源
System.out.println("创建文件夹成功.....");
}
/**
* 在HDFS上删除一个文件夹
*
* **/
public static void deleteDirectoryOnHDFS()throws Exception{
FileSystem fs=FileSystem.get(conf);
Path p =new Path("hdfs://10.2.143.5:9090/root/myfile");
fs.deleteOnExit(p);
fs.close();//释放资源
System.out.println("删除文件夹成功.....");
}
/**
* 在HDFS上创建一个文件
*
* **/
public static void createFileOnHDFS()throws Exception{
FileSystem fs=FileSystem.get(conf);
Path p =new Path("hdfs://10.2.143.5:9090/root/abc.txt");
fs.createNewFile(p);
//fs.create(p);
fs.close();//释放资源
System.out.println("创建文件成功.....");
}
/**
* 在HDFS上删除一个文件
*
* **/
public static void deleteFileOnHDFS()throws Exception{
FileSystem fs=FileSystem.get(conf);
Path p =new Path("hdfs://10.2.143.5:9090/root/abc.txt");
fs.deleteOnExit(p);
fs.close();//释放资源
System.out.println("删除成功.....");
}
/***
* 上传本地文件到
* HDFS上
*
* **/
public static void uploadFile()throws Exception{
//加载默认配置
FileSystem fs=FileSystem.get(conf);
//本地文件
Path src =new Path("D:\\6");
//HDFS为止
Path dst =new Path("hdfs://10.2.143.5:9090/root/");
try {
fs.copyFromLocalFile(src, dst);
} catch (IOException e) {
// TODO Auto-generated catch block
e.printStackTrace();
}
System.out.println("上传成功........");
fs.close();//释放资源
}
}

package com.java.api.hdfs;

import java.io.BufferedReader;
import java.io.IOException;
import java.io.InputStream;
import java.io.InputStreamReader;

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


/**
 * @author 三劫散仙
 * Java API操作HDFS
 * 工具类
 * 
 * **/
public class OperaHDFS {
	
	
	public static void main(String[] args)throws Exception {
		
		//System.out.println("aaa");
	//	uploadFile();
		//createFileOnHDFS();
		//deleteFileOnHDFS();
		//createDirectoryOnHDFS();
		//deleteDirectoryOnHDFS();
	//	renameFileOrDirectoryOnHDFS();
		//downloadFileorDirectoryOnHDFS();
		readHDFSListAll();
	}
	
	
	
	
	/***
	 * 加载配置文件
	 * **/
	 static Configuration conf=new Configuration();
	 
	 
	 
	 /**
	  * 重名名一个文件夹或者文件
	  * 
	  * **/
	 public static void renameFileOrDirectoryOnHDFS()throws Exception{
		 
			FileSystem fs=FileSystem.get(conf);
		    Path p1 =new Path("hdfs://10.2.143.5:9090/root/myfile/my.txt");
		    Path p2 =new Path("hdfs://10.2.143.5:9090/root/myfile/my2.txt");
			fs.rename(p1, p2);
			
			fs.close();//释放资源
			System.out.println("重命名文件夹或文件成功.....");
		 
	 }
	 
	 
	 /***
	  *  
	  * 读取HDFS某个文件夹的所有
	  * 文件，并打印
	  * 
	  * **/
	    public static void readHDFSListAll() throws Exception{
	    
	 //流读入和写入
	    	InputStream in=null;
	    	//获取HDFS的conf
	      //读取HDFS上的文件系统
	        FileSystem hdfs=FileSystem.get(conf);
	      //使用缓冲流，进行按行读取的功能
	        BufferedReader buff=null;
	      //获取日志文件的根目录
	        Path listf =new Path("hdfs://10.2.143.5:9090/root/myfile/");
	      //获取根目录下的所有2级子文件目录
	        FileStatus stats[]=hdfs.listStatus(listf);
	      //自定义j，方便查看插入信息
	        int j=0;
	         for(int i = 0; i < stats.length; i++){
	        	//获取子目录下的文件路径
	        	FileStatus   temp[]=hdfs.listStatus(new Path(stats[i].getPath().toString()));
	        	  for(int k = 0; k < temp.length;k++){
	        		  System.out.println("文件路径名:"+temp[k].getPath().toString());
	        	//获取Path
	        	Path p=new Path(temp[k].getPath().toString());
	        	//打开文件流
	        	 in=hdfs.open(p);
	        	 //BufferedReader包装一个流
	        	   buff=new BufferedReader(new InputStreamReader(in));	       	 
	        	 String str=null;
	             while((str=buff.readLine())!=null){
	            	 
	            	 System.out.println(str);
	             }
	                buff.close();
	                in.close();
	        	 
 
	        	 }
	        	 
	        	
	        	 
 
	        	  }
	        	 
	         hdfs.close();
	     

	    }
	 /**
	  * 从HDFS上下载文件或文件夹到本地
	  * 
	  * **/
	 public static void downloadFileorDirectoryOnHDFS()throws Exception{
		 
			FileSystem fs=FileSystem.get(conf);
		    Path p1 =new Path("hdfs://10.2.143.5:9090/root/myfile//my2.txt");
		    Path p2 =new Path("D://7");
			fs.copyToLocalFile(p1, p2);
			fs.close();//释放资源
			System.out.println("下载文件夹或文件成功.....");
		 
	 }
	 /**
	  * 在HDFS上创建一个文件夹
	  * 
	  * **/
	 public static void createDirectoryOnHDFS()throws Exception{
		 
			FileSystem fs=FileSystem.get(conf);
		    Path p =new Path("hdfs://10.2.143.5:9090/root/myfile");
			fs.mkdirs(p);
			fs.close();//释放资源
			System.out.println("创建文件夹成功.....");
		 
	 }
	 
	 /**
	  * 在HDFS上删除一个文件夹
	  * 
	  * **/
	 public static void deleteDirectoryOnHDFS()throws Exception{
		 
			FileSystem fs=FileSystem.get(conf);
		    Path p =new Path("hdfs://10.2.143.5:9090/root/myfile");
			fs.deleteOnExit(p);
			fs.close();//释放资源
			System.out.println("删除文件夹成功.....");
		 
	 }
	 /**
	  * 在HDFS上创建一个文件
	  * 
	  * **/
	 public static void createFileOnHDFS()throws Exception{
		 
			FileSystem fs=FileSystem.get(conf);
		    Path p =new Path("hdfs://10.2.143.5:9090/root/abc.txt");
			fs.createNewFile(p);
			//fs.create(p);
			fs.close();//释放资源
			System.out.println("创建文件成功.....");
		 
	 }
	 
	 /**
	  * 在HDFS上删除一个文件
	  * 
	  * **/
	 public static void deleteFileOnHDFS()throws Exception{
		 
			FileSystem fs=FileSystem.get(conf);
		    Path p =new Path("hdfs://10.2.143.5:9090/root/abc.txt");
			fs.deleteOnExit(p);
			fs.close();//释放资源
			System.out.println("删除成功.....");
		 
	 }
	 
	 
	/***
	 * 上传本地文件到
	 * HDFS上
	 * 
	 * **/
	public static void uploadFile()throws Exception{
		//加载默认配置
		FileSystem fs=FileSystem.get(conf);
		  //本地文件
        Path src =new Path("D:\\6");
        //HDFS为止
        Path dst =new Path("hdfs://10.2.143.5:9090/root/");
        try {
			fs.copyFromLocalFile(src, dst);
		} catch (IOException e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		}
        System.out.println("上传成功........");
   
        fs.close();//释放资源
 
		
	}

}

散仙用的是Hadoop1.2的版本，采用hadoop的eclipse插件在eclipse下编写调试的，下面给出截图：

下面给出一个读取某个文件夹下的所有的文件并打印内容的截图：

其实，java操作HDFS还是比较简单的，只要熟悉java IO的一些操作就可以了。

分享到：

怎么编译hadoop2.x的eclipse插件 | Hadoop确实是处理海量离线数据的利器

2016-01-24 11:21
浏览 462
评论(0)
分类:企业架构
查看更多

发表评论

您还没有登录,请您登录后再发表评论

最近访客更多访客>>

博主相关

文章分类

社区版块

存档分类

最新评论

HDFS

评论

发表评论

相关推荐

最近访客 更多访客>>

博主相关

文章分类

社区版块

存档分类

最新评论

HDFS

评论

发表评论

相关推荐

CDH与原生态hadoop之间的区别

Cloudera的CDH和Apache的Hadoop的区别

大数据、云计算系统高级架构师课程学习路线图

Oozie简介

清理ambari安装的hadoop集群

hawk大数据基础知识总结（2）

hawk大数据基础知识总结（1）

ambari是什么

数据仓库中的Inmon与Kimball架构之争

Hive的meta 数据支持以下三种存储方式

大数据要学习知识

Spark Streaming简介

pentaho套件

Impala：新一代开源大数据分析引擎

Weka是什么

解密Apache HAWQ ——功能强大的SQL-on-Hadoop引擎

Kettle的使用

clouder manager端口7180没有打开为什么

Impala与Hive的比较

Cloudera Manager、CDH零基础入门、线路指导

最近访客更多访客>>