package com.hdfs;


import java.io.FileOutputStream;
import java.io.OutputStream;


import org.apache.hadoop.fs.FSDataInputStream;
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 HdfsWork2 {
public static void main(String[] args) throws Exception {
myMerge();
}

public static void myWrite(Path path,FileSystem fs,OutputStream fsdos) throws Exception{
FileStatus[] fst = fs.listStatus(path);
FSDataInputStream fsdis = null;
for(int i=0;i<fst.length;i++){
if(fst[i].isDirectory()){
myWrite(fst[i].getPath(),fs,fsdos);
} else {
fsdis = fs.open(fst[i].getPath());
int read = 0;
byte[] buffer = new byte[255];
while((read=fsdis.read(buffer))>0){
fsdos.write(buffer, 0, read);
}
IOUtils.closeStream(fsdis);
}
}
}

public static void myMerge() throws Exception{
FileSystem fs = MyUtils.getFileSystem();
LocalFileSystem lfs = MyUtils.getLocalFileSystem();
Path localPath = new Path("D:/data/file0913.txt");
Path hdfsPath = new Path("/data2");

//FSDataOutputStream fsdos = lfs.create(localPath); 
//用上面的，報空指標異常，是因為系統原因，改成普通的檔案輸出流就好了
FileOutputStream fsdos = new FileOutputStream(localPath.toString());
myWrite(hdfsPath,fs,fsdos);
IOUtils.closeStream(fsdos);
}
}