Hadoop Java Hdfs API

1. 在本地文件系统生成一个文本文件，，读入文件，将其第101-120字节的内容写入HDFS成为一个新文件
2. 在HDFS中生成文本文件，读入这个文件，将其第101-120字节的内容写入本地文件系统成为一个新文件

环境部署：http://www.cnblogs.com/dopeter/p/4630791.html

FileBuilder.java

生成文件的工具类，包含在本地生成文件，在Hadoop生成文件，读取Hadoop指定目录的文件

package story;

import java.io.ByteArrayInputStream;
import java.io.ByteArrayOutputStream;
import java.io.FileNotFoundException;
import java.io.FileWriter;
import java.io.IOException;
import java.io.InputStream;
import java.io.OutputStream;
import java.io.PrintWriter;
import java.io.UnsupportedEncodingException;
import java.net.URI;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.IOUtils;
import org.apache.hadoop.util.Progressable;

public class FileBuilder {
    
    //build default test data
    public static String BuildTestFileContent()
    {
        StringBuilder contentBuilder=new StringBuilder();
        
        for(int loop=0;loop<100;loop++)
            contentBuilder.append(String.valueOf(loop));
        
        String content =contentBuilder.toString();
        
        return content;
    }    
    
    //build local file
    public static void BuildLocalFile(String buildPath,String content) throws FileNotFoundException, UnsupportedEncodingException
    {        
        /*
        FileWriter fileWriter;
        try {
            fileWriter = new FileWriter(buildPath);
            
            fileWriter.write(content);        
            fileWriter.close();
        } catch (IOException e) {            
            e.printStackTrace();
        }
        */
        


            PrintWriter out = new java.io.PrintWriter(new java.io.File(buildPath), "UTF-8");
            String text = new java.lang.String(content);
            out.print(text);
            out.flush();
            out.close();

    }
    
    //upload file to hadoop
    public static void BuildHdfsFile(String buildPath,byte[] fileContent) throws IOException
    {
        //convert to inputstream
        InputStream inputStream=new ByteArrayInputStream(fileContent); 
        
        //hdfs upload
        Configuration conf = new Configuration();
        
        FileSystem fs = FileSystem.get(URI.create(buildPath), conf);
        OutputStream outputStream = fs.create(new Path(buildPath), new Progressable() {
            public void progress() {
                System.out.print(".");        
            }
        });
        
        IOUtils.copyBytes(inputStream, outputStream, fileContent.length, true);
    }
    
    //wrapper for upload file
    public static void BuildHdfsFile(String buildPath,String fileContent) throws IOException
    {
        BuildHdfsFile(buildPath,fileContent.getBytes());
    }
    
    //download file from hadoop
    public static byte[] ReadHdfsFile(String readPath)throws IOException
    {
        byte[] fileBuffer;
        Configuration conf = new Configuration();
        FileSystem fs = FileSystem.get(URI.create(readPath), conf);
        InputStream in = null;
        ByteArrayOutputStream out=new ByteArrayOutputStream();
        try {
        in = fs.open(new Path(readPath));
        IOUtils.copyBytes(in, out, 4096, false);
        
        fileBuffer=out.toByteArray();
        } finally {
        IOUtils.closeStream(in);
        }
        
        return fileBuffer;
    }
    
}

FileContentHandler.java

文件内容的处理类，读取本地文件时设置起始Position与截取的长度，读取从Hadoop下载的文件时设置起始Position与截取的长度

package story;

import java.io.IOException;
import java.io.RandomAccessFile;
import java.io.UnsupportedEncodingException;

public class FileContentHandler {
    public static byte[] GetContentByLocalFile(String filePath,long beginPosition,int readLength)
    {
        int readBufferSize=readLength;
        byte[] readBuffer=new byte[readBufferSize];
        
        RandomAccessFile  accessFile;
        try {
            accessFile=new RandomAccessFile (filePath,"r");
            long length=accessFile.length();
            System.out.println(length);
            
            if(length>beginPosition&&length>beginPosition+readBufferSize)
            {
                accessFile.seek(beginPosition);
                accessFile.read(readBuffer);
                accessFile.close();
            }
        } catch ( IOException e) {
            // TODO Auto-generated catch block
            e.printStackTrace();
        }
        
        return readBuffer;
    }
    
    public static String GetContentByBuffer(byte[] buffer,int beginPosition,int readLength) throws UnsupportedEncodingException
    {
        String content;
        byte[] subBuffer=new byte[readLength];
        for(int position=0;position<readLength;position++)
            subBuffer[position]=buffer[beginPosition+position];
        
        buffer=null;
        
        content=new String(subBuffer,"UTF-8");
        System.out.println(content);
        
        return content;
    }
    
}

UploadStory.java

1的流程代码

package story;

public class UploadStory {
    
    //public static void main(String[] args) throws Exception {}
    
    public static void main(String[] args) throws Exception {        
        //also define value of parameter from arguments.
        String localFilePath="F:/bulid.txt";
        String hdfsFilePath="hdfs://hmaster0:9000/user/14699_000/input/build.txt";
        int readBufferSize=20;
        long fileBeginReadPosition=101;
        
        //upload story begin.
        
        //build local file        
        FileBuilder.BuildLocalFile(localFilePath,FileBuilder.BuildTestFileContent());        
        //read file        
        byte[] uploadBuffer=FileContentHandler.GetContentByLocalFile(localFilePath, fileBeginReadPosition, readBufferSize);    
        //upload 
        if(uploadBuffer!=null&&uploadBuffer.length>0)        
            FileBuilder.BuildHdfsFile(hdfsFilePath, uploadBuffer);

    }
    
}

DownloadStory.java

2的流程代码

package story;

public class DownloadStory {
    
    //public static void main(String[] args) throws Exception {        }
    
    
    public static void main(String[] args) throws Exception {        
        //also define value of parameter from arguments.
        String localFilePath="F:/bulid.txt";
        String hdfsFilePath="hdfs://hmaster0:9000/user/14699_000/input/build2.txt";
        int readBufferSize=20;
        int fileBeginReadPosition=101;
        
        //build file to hadoop
        FileBuilder.BuildHdfsFile(hdfsFilePath, FileBuilder.BuildTestFileContent());
        
        //download file 
        byte[] readBuffer=FileBuilder.ReadHdfsFile(hdfsFilePath);
        
        //handle buffer
        String content=FileContentBuilder.GetContentByBuffer(readBuffer, fileBeginReadPosition, readBufferSize);
        
        //write to local file
        FileBuilder.BuildLocalFile(localFilePath, content);
    }
    
}