本地多级文件 合并上传到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 HdfsWork1 {
public static void main(String[] args) throws Exception {
myMerge();
}

public static void myWrite(Path path,LocalFileSystem lfs,FSDataOutputStream fsdos) throws Exception{
FileStatus[] fst = lfs.listStatus(path);//获得本地文件的数组
FSDataInputStream fsdis = null;
for(int i=0;i<fst.length;i++){
if(fst[i].isDirectory()){//是文件夹继续调用本方法
myWrite(fst[i].getPath(),lfs,fsdos);
} else {
fsdis = lfs.open(fst[i].getPath());//打开文件输入流
int read = 0;
byte[] buffer = new byte[255];
while((read=fsdis.read(buffer))>0){//读取文件并输出到hdfs上
fsdos.write(buffer, 0, read);
}
IOUtils.closeStream(fsdis);//关闭文件输入流
}
}
}

public static void myMerge() throws Exception{
FileSystem fs = MyUtils.getFileSystem();//获取hdfs文件系统
LocalFileSystem lfs = MyUtils.getLocalFileSystem();//获得本地文件系统
Path localPath = new Path("D:/data");//本地路径
Path hdfsPath = new Path("/data/file1.txt");//hdfs路径

FSDataOutputStream fsdos = fs.create(hdfsPath);//dfs文件输出流
myWrite(localPath,lfs,fsdos);//调用递归方法
IOUtils.closeStream(fsdos);//关闭输出流
}
}

下面是工具类

package com.utils;

import org.apache.hadoop.conf.Configuration;

import org.apache.hadoop.fs.FileSystem;

import org.apache.hadoop.fs.LocalFileSystem;

/**

 * 

 * @described 获取文件系统

 * @author 

 * @date 

 * @see 

 */

public class MyUtils {

public static FileSystem getFileSystem() throws Exception{
Configuration conf =new Configuration();
FileSystem fs = FileSystem.newInstance(conf);
return fs;

}
public static LocalFileSystem getLocalFileSystem() throws Exception{
return FileSystem.newInstanceLocal(new Configuration());
}

}