本地多级文件 原样上传到hdfs

package com.hdfs;

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.LocalFileSystem;

import org.apache.hadoop.fs.Path;

import org.apache.hadoop.io.IOUtils;

import com.beicai.utils.MyUtils;

/**

 * 

 * @described 本地多级文件 原样上传到hdfs

 */

public class HdfsWork {

public static void main(String[] args) throws Exception {
myMerge();
System.out.println("ok");
}

public static void myWrite(Path path,LocalFileSystem lfs,FileSystem fs) throws Exception{
FileStatus[] fst = lfs.listStatus(path);
for(int i=0;i<fst.length;i++){
Path dir = fst[i].getPath();//获取路径
//file:/D:/data/a
dir = new Path(dir.toString().split(":")[2]);//截取相对本地磁盘D盘的绝对路径

if(fst[i].isDirectory()){//是文件夹

myWrite(fst[i].getPath(),lfs,fs);//调用本身
fs.mkdirs(dir);//创建里面没有文件的文件夹（空文件夹）
} else {
FSDataInputStream fsdis = lfs.open(fst[i].getPath()); //打开文件输入流
FSDataOutputStream fsdos = fs.create(dir);//打开文件输出流流

int read = 0;
byte[] buffer = new byte[255];
while((read=fsdis.read(buffer))>0){
fsdos.write(buffer, 0, read);
}

IOUtils.closeStream(fsdis);//关闭流
IOUtils.closeStream(fsdos);
}
}
}

public static void myMerge() throws Exception{
FileSystem fs = MyUtils.getFileSystem();
LocalFileSystem lfs = MyUtils.getLocalFileSystem();
Path localPath = new Path("D:/datas");//本地路径

myWrite(localPath,lfs,fs);//调用方法
}

}