使用HDFS 简单实现云盘系统

Posted 2022-01-08 Maynor大数据

HDFS作为Hadoop的文件系统，具有以下特点：

高容错
流式文件访问
适合大数据处理
适合批处理
高容错率可以使它构建在廉洁的机器上
当然也有缺点：

不擅长低延迟和高吞吐率的数据访问
小文件存储
并发写入和高频率修改
参照：https://blog.csdn.net/woshiwanxin102213/article/details/19990487

HDFS常用指令
HDFS的指令很多都与Linux有相同之处。官网指令参考地址
指出几个不同且常用的：

hadoop fs -get #将文件从HDFS文件系统中拷贝到本地文件系统中
hadoop fs -put # 将本地文件系统中的文件拷贝到HDFS文件系统中

剩下的 ls，cd， cp， mkdir等操作基本与Linux的指令相同，如操作的是HDFS系统只需要 hadoop fs -指令 就 ok了。

HDFS Java API
官网API文档地址
我们不可能一次性把所有的API全部学完，只能先常用的，其它的则在需要的时候在去查文档。所有查文档很重要。

这里以我自己的一个基于HDFS的简单云盘练习项目做介绍：

整体文件目录结构，eclipse中的maven WebApp 框架
（这里报错是因为我在maven仓库中添加了一个自定义的包与maven有冲突，但四没有关西，项目能够正常运行）

服务端：

web前端：

创建HDFS连接
在src/main/resources目录中加入Hadoop配置文件中的core-site.xml 和 hdfs-site.xml。

在Dao层中编写一个连接器

public class HdfsDao 
	private static Configuration conf;
	static
		System.setProperty("HADOOP_USER_NAME", "root"); //（1）
		conf = new Configuration();  //（2）
	
	

	public static FileSystem getFileSystem() throws IOException
		return FileSystem.get(conf);
	

（1）指定访问HDFS时的用户名，如果不指定则为当前电脑的用户名。
（2）读取resources中的配置文件。

API操作 业务层
这里只提供一些HDFS API操作

public class HdfsBizImpl 
// 重命名文件
public boolean rename(String oldName, String newName, String currentPath) throws IOException
	Path src = new Path(currentPath + oldName);
	Path dst = new Path(currentPath + newName);
	boolean result = HdfsDao.getFileSystem().rename(src, dst);
	return result;

​

		/**
 * 查询当前路径下得所有文件
 * @param path
 * @return
 * @throws FileNotFoundException
 * @throws IOException
 */
public List<HdfsFile> getFileList(String path) throws FileNotFoundException, IOException
	Path root = new Path(path);
	FileStatus[] fs = HdfsDao.getFileSystem().listStatus(root);
	
	List<HdfsFile> list = getFiles( fs);
	return list;


// 将读取到的文件信息封装成HdfsFile对象
private List<HdfsFile> getFiles(FileStatus[] fs)
	List<HdfsFile> list = new ArrayList<>();
	for(FileStatus lfs : fs)
		HdfsFile hf = new HdfsFile();
		Path p = lfs.getPath();
		String fileName = p.getName();
		hf.setFileName(fileName);
		
		boolean b = lfs.isFile();
		hf.setFile(b);
		
		Long len = null;
		String sizeString = null;
		if ( b )
			len = lfs.getLen();
			sizeString = formatSize(len);
			hf.setSizeString(sizeString);
		
		long mt = lfs.getModificationTime();
		Date d = new Date(mt);
		DateFormat df = new SimpleDateFormat("yyyy年MM月dd日 HH:mm:ss");
		String dateString = df.format( d );
		hf.setDateString(dateString);
		hf.setFilePath(lfs.getPath().toString());
		list.add(hf);
	
	return list;


// 将字节数据格式化
private String formatSize(long len)
	String result = "";
	if (len/1024/1024/1024 > 0)
		result = len/1024/1024/1024 + "G";
	else if (len/1024/1024 > 0)
		result = len/1024/1024 + "M";
	else if (len/1024 > 0)
		result = len/1024 + "K";
	else
		result = len + "B";
	
	return result;


// 删除某个文件
public boolean delFile(String oldName, String currentPath) throws IOException 
	Path path = new Path(currentPath + oldName);
	boolean result = HdfsDao.getFileSystem().delete(path,true);
	
	return result;


// 创建目录
public boolean mkdir(String newName, String currentPath) throws IOException 
	Path f = new Path(currentPath + newName);
	boolean result = HdfsDao.getFileSystem().mkdirs(f);
	return result;


//上传文件
public boolean uploadFile(Map<String, String> map) 
	String realPath = map.get("realPath");
	String currentPath = map.get("currentPath");
	String fileName = map.get("fileName");
	
	if (!currentPath.endsWith("/"))
		currentPath += "/";
	
	
	Path src = new Path(realPath);
	Path dst = new Path(currentPath + fileName);
	
	try 
		HdfsDao.getFileSystem().copyFromLocalFile(src, dst);
		File f = new File(realPath);
		if (f.exists())
			f.delete();
		
		return true;
	 catch (IOException e) 
		e.printStackTrace();
	
	return false;


// 移动文件
public boolean moveTo(String oldName, String newPath, String currentPath) throws AccessControlException, FileAlreadyExistsException, FileNotFoundException, ParentNotDirectoryException, UnsupportedFileSystemException, IOException 
	Path src = new Path(currentPath + oldName);
	Path dst = new Path(newPath + oldName);
	boolean result = FileContext.getFileContext().util().copy(src, dst, true, false);
	return result;


// 复制文件
public boolean copyTo(String oldName, String newPath, String currentPath) throws AccessControlException, FileAlreadyExistsException, FileNotFoundException, ParentNotDirectoryException, UnsupportedFileSystemException, IOException 
	Path src = new Path(currentPath + oldName);
	Path dst = new Path(newPath + oldName);
	boolean result = FileContext.getFileContext().util().copy(src, dst);
	return result;
	


// 在当前目录下查找文件
public List<HdfsFile> selectFile(String currentPath, String type,Map<String,Set<String>> map) throws AccessControlException, FileNotFoundException, UnsupportedFileSystemException, IllegalArgumentException, IOException 
	RemoteIterator<LocatedFileStatus> ri = HdfsDao.getFileSystem().listFiles(new Path(currentPath), true);
	List<HdfsFile> list = new ArrayList<>();
	
	while(ri.hasNext())
		LocatedFileStatus lfs = ri.next();
		String fileName = lfs.getPath().getName();
		if (map != null)
			if ("0".equals(type))
				addList(list, lfs);
			else
				Set<String> set = map.get(type);
				for (String ext : set) 
					if (fileName.endsWith(ext))
						addList(list, lfs);
					
				
			
		
	
	
	return list;



private void addList(List<HdfsFile> list, LocatedFileStatus lfs) 
	HdfsFile hf = new HdfsFile();
	Path p = lfs.getPath();
	String fileName = p.getName();
	hf.setFileName(fileName);
	
	boolean b = lfs.isFile();
	hf.setFile(b);
	
	Long len = null;
	String sizeString = null;
	if ( b )
		len = lfs.getLen();
		sizeString = formatSize(len);
		hf.setSizeString(sizeString);
	
	long mt = lfs.getModificationTime();
	Date d = new Date(mt);
	DateFormat df = new SimpleDateFormat("yyyy年MM月dd日 HH:mm:ss");
	String dateString = df.format( d );
	hf.setDateString(dateString);
	hf.setFilePath(lfs.getPath().toString());
	list.add(hf);


// 下载文件到本地
public String downloadFile(String currentPath, String fileName, String local) throws Exception 
	String src = currentPath + fileName;
	String dest = local + fileName;
	
	Path p = new Path(src);
	if ( !HdfsDao.getFileSystem().isFile(p))
		throw new Exception(src + " is not a valid file");
	
	
	HdfsDao.getFileSystem().copyToLocalFile(p, new Path(dest));
	return dest;

​

运行效果