从本地上传整个目录到hdfs的java程序

• Z时代
• 2024-01-10
• 分类：综合

首先在网上找了好久没有找到从本地文件系统上传整个目录到hdfs文件系统的程序，权威指南上也没有，都是单个文件上传，所以这里自己编写了一个程序，封装成jar包执行能够复制。

先说明一下代码：须要手动输入两个路径，一个本地文件/目录路径，第二个是hdfs目录路径。好直接上代码：

import java.io.BufferedInputStream;
import java.io.File;
import java.io.FileInputStream;
import java.io.InputStream;
import java.io.OutputStream;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileStatus;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.IOUtils;
import org.apache.hadoop.util.Progressable;
/**
 * 
 * @author zlqiao
 *
 */
public class Copy {
	public static void main(String[] args) throws Exception {
		if(args.length < 2){
			System.out.println("Please input two number");
			System.exit(2);
		}
		String localSrc = args[0];
		String dst = args[1];
		Configuration conf = new Configuration();
		File srcFile = new File(localSrc);
		if(srcFile.isDirectory()){
			copyDirectory(localSrc , dst , conf);
		}else{
			copyFile(localSrc, dst, conf);
		}
	}
	/**
	 * 复制文件
	 * @param src
	 * @param dst
	 * @param conf
	 * @return
	 * @throws Exception
	 */
	public static boolean copyFile(String src , String dst , Configuration conf) throws Exception{
		FileSystem fs = FileSystem.get(conf);
		fs.exists(new Path(dst));
		//FileStatus status = fs.getFileStatus(new Path(dst));
		File file = new File(src);
		InputStream in = new BufferedInputStream(new FileInputStream(file));
		/**
		 * FieSystem的create方法能够为文件不存在的父文件夹进行创建，
		 */
		OutputStream out = fs.create(new Path(dst) , new Progressable() {
			public void progress() {
				System.out.print(".");
			}
		});
		IOUtils.copyBytes(in, out, 4096, true);
		return true;
	}
	/**
	 * 复制文件夹
	 * @param src
	 * @param dst
	 * @param conf
	 * @return
	 * @throws Exception 
	 */
	public static boolean copyDirectory(String src , String dst , Configuration conf) throws Exception{
		FileSystem fs = FileSystem.get(conf);
		if(!fs.exists(new Path(dst))){
			fs.mkdirs(new Path(dst));
		}
		System.out.println("copyDirectory:"+dst);
		FileStatus status = fs.getFileStatus(new Path(dst));
		File file = new File(src);
		if(status.isFile()){
			System.exit(2);
			System.out.println("You put in the "+dst + "is file !");
		}else{
			dst = cutDir(dst);
		}
		File[] files = file.listFiles();
		for(int i = 0 ;i< files.length; i ++){
			File f = files[i];
			if(f.isDirectory()){
				copyDirectory(f.getPath(),dst,conf);
			}else{
				copyFile(f.getPath(),dst+files[i].getName(),conf);
			}
		}
		return true;
	}
	public static String cutDir(String str){
		String[] strs = str.split(File.pathSeparator);
		String result = "";
		if("hdfs"==strs[0]){
			result += "hdfs://";
			for(int i = 1 ; i < strs.length  ; i++){
				result += strs[i] + File.separator;
			}
		}else{
			for(int i = 0 ; i < strs.length  ; i++){
				result += strs[i] + File.separator;
			}
		}
		return result;
	}
}

以上是 从本地上传整个目录到hdfs的java程序 的全部内容， 来源链接： utcz.com/z/390141.html