Hadoop之操作HDFS示例

1.引入pom文件

    <dependency>
            <groupId>org.apache.hadoop</groupId>
            <artifactId>hadoop-client</artifactId>
            <version>2.7.7</version>
        </dependency>
        <dependency>
            <groupId>junit</groupId>
            <artifactId>junit</artifactId>
            <version>LATEST</version>
        </dependency>

JavaApi对HDFS进行文件上传删除 重命名等操作

import org.apache.hadoop.conf.Configuration;

import org.apache.hadoop.fs.*;

import org.apache.hadoop.fs.permission.FsAction;
import org.apache.hadoop.fs.permission.FsPermission;
import org.apache.hadoop.io.IOUtils;
import org.apache.hadoop.util.Progressable;
import org.junit.After;
import org.junit.Before;
import org.junit.Test;

import java.io.*;
import java.net.URI;


public class FileSystemUtil {

    private static final  String   HDFS_URL="hdfs://192.168.80.153:8020";
    private static final  String HDFS_USER="root";
    private static FileSystem fileSystem;

    @Before
    public  void  prepare(){
        try {
            Configuration configuration=new Configuration(  );
            configuration.set( "dfs.replication","1" );
            fileSystem= (FileSystem) FileSystem.get( new URI(HDFS_URL),configuration,HDFS_USER );
        }catch (Exception ex){
            ex.printStackTrace();
        }
    }

    @After
    public void distory(){
        fileSystem=null;
    }


    //在HDFS创建目录
    //目录支持级联创建
    @Test
    public void mkDir() throws Exception {
        fileSystem.mkdirs(new Path("/hdfs-api/test0/"));
    }

    //创建目录并给目录设置对应的权限
    //第一个参数表示 创建者用户的读写权限
    //第二个参数表示  创建者同组用户权限
    //第三个参数表示 其他用户权限
    @Test
    public void mkdirWritePermission() throws IOException {
        fileSystem.mkdirs(  new Path( "/hdfs-api/test1" ),
                new FsPermission( FsAction.READ_WRITE,FsAction.READ,FsAction.READ ) );
    }

    //创建文件夹写入内容
    @Test
    public void mkdirAndWriteFile() throws IOException {
        //第一个参数文件所在hdfsurl的地址
        //第二个参数 控制文件如果存在是否覆盖
        //第三个参数 控制文件的缓冲区大小
        FSDataOutputStream fsDataOutputStream= fileSystem.create(
                new Path(  "/hdfs-api/test0/example.txt"),
                true,4096);
        fsDataOutputStream.write( "hello,hadoop!".getBytes());
        fsDataOutputStream.write( "hello world!".getBytes() );
        //强制缓冲区的内容输出
        fsDataOutputStream.flush();
        fsDataOutputStream.close();
    }

    //判断文件是否存在
    @Test
    public void existsFile() throws IOException {
        boolean exists = fileSystem.exists( new Path( "/hdfs-api/test0/example.txt" ) );
        System.out.println(exists);
    }

    //读取文件内容
    @Test
    public void readToString() throws IOException {
        FSDataInputStream fsDataInputStream=fileSystem.open( new Path( "/hdfs-api/test1/example.txt" )) ;
        BufferedReader bufferedReader=new BufferedReader( new InputStreamReader( fsDataInputStream ,"UTF-8"));
        StringBuilder sb=new StringBuilder(  );
        String str="";
        while(( str=bufferedReader.readLine())!=null){
            sb.append( str);
        }
        System.out.println(sb.toString());

    }
    //重命名文件
    @Test
    public void renameFile() throws IOException {
        Path oldPath=new Path( "/hdfs-api/test1/example.txt" );
        Path newPath=new Path( "/hdfs-api/test1/example_new.txt" );
        boolean rename = fileSystem.rename( oldPath, newPath );
        System.out.println(rename);
    }

    //删除文件的时候权限不够也会删除失败但是不会抛出异常
    @Test
    public void delectFile()  {
        //第一个参数 文件地址 第二个参数表示是否递归删除如果path是一个目录
        // true则删除该目录中所有文件
        // false 如果目录下面有文件则抛出异常
        try {
            boolean delete = fileSystem.delete( new Path( "/centos7.ios" ), true );
            System.out.println( delete );
        }catch (Exception ex){
            ex.printStackTrace();
        }
    }

    //上传文件
    @Test
    public void uploadFile() throws IOException {
        Path src=new Path( "C:\\Users\\reality\\Desktop\\Api\\hdfsexample\\src\\main\\java\\FileSystemUtil.java" );
        Path upsrc=new Path( "/" );
        fileSystem.copyFromLocalFile( src,upsrc );
    }

    //上传大文件并且显示进度
    @Test
    public void compyFromLocalBigFile() throws IOException {
        File file=new File("D:\\CentOS-7-x86_64-DVD-1908.iso");
        final float fileSize=file.length();
        InputStream inputStream=new BufferedInputStream( new FileInputStream( file ) );
        FSDataOutputStream out=fileSystem.create( new Path( "/centos7.ios" ),
                //自定义显示进度
                new Progressable() {
                    long fileCount=0;
                    public void progress() {
                        fileCount++;
                        System.out.println("上传进度"+(fileCount * 64 * 1024 / fileSize)*100+"%");
                    }
                }
        );
        IOUtils.copyBytes( inputStream,out,4096 );
    }


    //从HDFS下载文件
    @Test
    public void downOnload() throws IOException {
        Path src=new Path( "/workcount/input.txt" );
        Path downPath=new Path( "D:\\" );
        //第一个参数控制下载完成之后是否删除源文件
        //第二个参数HDFS地址
        //第三个参数表示 下载到本地的地址
        //第四个参数  useRawLocalFileSystem  是否用作本地系统因为有点时候可能会存在文件系统不兼容的情况下  Linux的文件
        //Windows下载经常出现错误
        fileSystem.copyToLocalFile( false,src,downPath,true );
    }
    //查看目录下文件的基本信息
    @Test
    public void listFiles() throws IOException {
        FileStatus[] fileStatuses = fileSystem.listStatus( new Path( "/" ));
        for (FileStatus fileStatus : fileStatuses) {
            //fileStatus的toString被重写过 所以可以看到存放位置 是否是文件等信息
            System.out.println(fileStatus.toString());
        }
    }

    //递归查看目录下所有文件的信息 比listStatus多出了文本大小，副本系数，块大小信息
    @Test
    public void listFilesRecursive() throws IOException {
        RemoteIterator <LocatedFileStatus> locatedFileStatusRemoteIterator = fileSystem.listFiles( new Path( "/" ), true );
        while(locatedFileStatusRemoteIterator.hasNext()){
            System.out.println(locatedFileStatusRemoteIterator.next());
        }
    }

    //查看文件的块信息
    //块信息包括文件起始偏移量 w文件大小 文件存放的位置  块所在的主机名
    @Test
    public void getFileBlockLocations() throws IOException {
        FileStatus fileStatus = fileSystem.getFileStatus( new Path( "/workcount/input.txt" ) );
        BlockLocation[] fileBlockLocations = fileSystem.getFileBlockLocations( fileStatus, 0, fileStatus.getLen() );
        for (BlockLocation fileBlockLocation : fileBlockLocations) {
            System.out.println(fileBlockLocation);
        }
    }
}