Hadoop（四）HDFS的高级API操作

一 HDFS客户端环境准备

1.1 jar包准备

1）解压hadoop-2.7.6.tar.gz到非中文目录

2）进入share文件夹，查找所有jar包，并把jar包拷贝到_lib文件夹下

3）在全部jar包中查找sources.jar，并剪切到_source文件夹。

4）在全部jar包中查找tests.jar，并剪切到_test文件夹

1.2 Eclipse准备

1）根据自己电脑的操作系统拷贝对应的编译后的hadoop jar包到非中文路径（例如：E:\02_software\hadoop-2.7.6）。（如果不生效，重新启动eclipse）

2）配置HADOOP_HOME环境变量

3）创建第一个java工程HdfsClientDemo1

4）创建lib文件夹，然后添加jar包

5）创建包，HdfsClient测试类

public class HdfsClient {

    // 上传文件

    public static void main(String[] args) throws IOException, InterruptedException, URISyntaxException {

        // 1 获取文件系统

        Configuration configuration = new Configuration();

        // 配置在集群上运行

        // configuration.set("fs.defaultFS", "hdfs://node21:9000");

        // FileSystem fs = FileSystem.get(configuration);

        FileSystem fs = FileSystem.get(new URI("hdfs://node21:9000"), configuration, "admin");

        // 2 上传文件

        fs.copyFromLocalFile(new Path("e:/hello.txt"), new Path("/hello2.txt"));

        // 3 关闭资源

        fs.close();

        System.out.println("over");

    }

}

6）执行程序

运行时需要配置用户名称，客户端去操作hdfs时，是有一个用户身份的。默认情况下，hdfs客户端api会从jvm中获取一个参数来作为自己的用户身份：-DHADOOP_USER_NAME=admin，admin为用户名称。

7）注意：如果eclipse打印不出日志，在控制台上只显示

1.log4j:WARN No appenders could be found for logger (org.apache.hadoop.util.Shell).  

2.log4j:WARN Please initialize the log4j system properly.  

3.log4j:WARN See http://logging.apache.org/log4j/1.2/faq.html#noconfig for more info.

需要在项目的src目录下，新建一个文件，命名为“log4j.properties”，在文件中填入

log4j.rootLogger=INFO, stdout

log4j.appender.stdout=org.apache.log4j.ConsoleAppender

log4j.appender.stdout.layout=org.apache.log4j.PatternLayout

log4j.appender.stdout.layout.ConversionPattern=%d %p [%c] - %m%n

log4j.appender.logfile=org.apache.log4j.FileAppender

log4j.appender.logfile.File=target/spring.log

log4j.appender.logfile.layout=org.apache.log4j.PatternLayout

log4j.appender.logfile.layout.ConversionPattern=%d %p [%c] - %m%n

二 HDFS的高级API编程

2.1 HDFS文件上传（测试参数优先级）

1.代码

@Test

public void testCopyFromLocalFile() throws IOException, InterruptedException, URISyntaxException {

// 1 获取文件系统

Configuration configuration = new Configuration();

configuration.set("dfs.replication", "2");

FileSystem fs = FileSystem.get(new URI("hdfs://node21:9000"), configuration, "admin");

// 2 上传文件

fs.copyFromLocalFile(new Path("e:/hello.txt"), new Path("/hello5.txt"));

// 3 关闭资源

fs.close();

System.out.println("over");

}

2.将hdfs-site.xml拷贝到项目的根目录下

<?xml version="1.0" encoding="UTF-8"?>

<?xml-stylesheet type="text/xsl" href="configuration.xsl"?>

<configuration>

  <property>

    <name>dfs.replication</name>

    <value>1</value>

  </property>

</configuration>

3.测试参数优先级

参数优先级：（1）客户端代码中设置的值 >（2）classpath下的用户自定义配置文件 >（3）然后是服务器的默认配置

2.2 文件的上传和下载

package com.xyg.hdfs.api;

import org.apache.hadoop.conf.Configuration;

import org.apache.hadoop.fs.FileSystem;

import org.apache.hadoop.fs.Path;

public class HDFS_GET_AND_PUT {

    public static void main(String[] args) throws Exception {

        Configuration conf = new Configuration();

        conf.set("fs.defaultFS", "hdfs://node21:9000");

        conf.set("dfs.replication", "2");

        FileSystem fs = FileSystem.get(conf);      

        /**

         * 更改操作用户有两种方式:

         * 1、直接设置运行换种的用户名为hadoop

         * VM arguments ;   -DHADOOP_USER_NAME=admin

         * 2、在代码中进行声明

         * System.setProperty("HADOOP_USER_NAME", "admin");

         */

        System.setProperty("HADOOP_USER_NAME", "admin");

        // 上传

        fs.copyFromLocalFile(new Path("c:/sss.txt"), new Path("/a/ggg.txt"));     

        /**

         * .crc  ： 校验文件

         * 每个块的元数据信息都只会记录合法数据的起始偏移量：  qqq.txt  blk_41838 :  0 - 1100byte

         * 如果进行非法的数据追加。最终是能够下载合法数据。

         * 由于你在数据的中间， 也就是说在 0 -1100 之间的范围进行了数据信息的更改。 造成了采用CRC算法计算出来校验值，和最初存入进HDFS的校验值

         * 不一致。HDFS就认为当前这个文件被损坏了。

         */

        // 下载

        fs.copyToLocalFile(new Path("/a/qqq.txt"), new Path("c:/qqq3.txt"));

        /**

         * 上传和下载的API的底层封装其实就是 ： FileUtil.copy(....)

         */

        fs.close();

    }

}

2、配置文件conf

package com.xyg.hdfs;

import java.io.IOException;

import java.util.Iterator;

import java.util.Map.Entry;

import org.apache.hadoop.conf.Configuration;

import org.apache.hadoop.fs.FileSystem;

public class TestConf1 {

    public static void main(String[] args) throws Exception {

        /**

         * 底层会加载一堆的配置文件：

         *

         * core-default.xml

         * hdfs-default.xml

         * mapred-default.xml

         * yarn-default.xml

         */

        Configuration conf = new Configuration();

//        conf.addResource("hdfs-default.xml");

        /**

         * 当前这个hdfs-site.xml文件就放置在这个项目中的src下。也就是classpath路径下。

         * 所以 FS在初始化的时候，会把hdfs-site.xml这个文件中的name-value对解析到conf中

         * 但是：

         * 1、如果hdfs-site.xml 不在src下， 看是否能加载？？？  不能

         * 2、如果文件名不叫做 hdfs-default.xml 或者 hdsf-site.xml  看是否能自动加载？？？  不能

         * 得出的结论：

         * 如果需要项目代码自动加载配置文件中的信息，那么就必须把配置文件改成-default.xml或者-site.xml的名称

         * 而且必须放置在src下

         *

         * 那如果不叫这个名，或者不在src下，也需要加载这些配置文件中的参数：

         * 必须使用conf对象提供的一些方法去手动加载

         */

//        conf.addResource("hdfs-site.xml");

        conf.set("dfs.replication", "1");

        conf.addResource("myconfig/hdfs-site.xml");

        /**

         * 依次加载的参数信息的顺序是：

         * 1、加载 core/hdfs/mapred/yarn-default.xml

         * 2、加载通过conf.addResources()加载的配置文件

         * 3、加载conf.set(name, value)

         */

        FileSystem fs = FileSystem.get(conf);

        System.out.println(conf.get("dfs.replication"));

        Iterator<Entry<String, String>> iterator = conf.iterator();

        while(iterator.hasNext()){

            Entry<String, String> e = iterator.next();

            System.out.println(e.getKey() + "\t" + e.getValue());

        }

    }

}

输出结果

3、列出指定目录下的文件以及块的信息

package com.xyg.hdfs;

import org.apache.hadoop.conf.Configuration;

import org.apache.hadoop.fs.BlockLocation;

import org.apache.hadoop.fs.FileSystem;

import org.apache.hadoop.fs.LocatedFileStatus;

import org.apache.hadoop.fs.Path;

import org.apache.hadoop.fs.RemoteIterator;

public class TestHDFS1 {

    public static void main(String[] args) throws Exception {

        Configuration conf = new Configuration();

        System.setProperty("HADOOP_USER_NAME", "admin");

        conf.set("fs.defaultFS", "hdfs://node21:9000");

        FileSystem fs = FileSystem.get(conf);

        /**

         * 列出指定的目录下的所有文件

         */

        RemoteIterator<LocatedFileStatus> listFiles = fs.listFiles(new Path("/"), true);

        while(listFiles.hasNext()){

            LocatedFileStatus file = listFiles.next();

            System.out.println(file.getPath()+"\t");

            System.out.println(file.getPath().getName()+"\t");

            System.out.println(file.getLen()+"\t");

            System.out.println(file.getReplication()+"\t");

            /**

             * blockLocations的长度是几？  是什么意义？

             *

             * 块的数量

             */

            BlockLocation[] blockLocations = file.getBlockLocations();

            System.out.println(blockLocations.length+"\t");

            for(BlockLocation bl : blockLocations){

                String[] hosts = bl.getHosts();

                System.out.print(hosts[0] + "-" + hosts[1]+"\t");

            }

            System.out.println();

        }

    }

}

输出结果

hdfs://hadoop1:9000/aa/bb/cc/hadoop.tar.gz

hadoop.tar.gz

199007110

2

3

hadoop3-hadoop1    hadoop1-hadoop2    hadoop1-hadoop4

4、上传文件

package com.xyg.hdfs;

import java.io.File;

import java.io.FileInputStream;

import java.io.InputStream;

import org.apache.hadoop.conf.Configuration;

import org.apache.hadoop.fs.FSDataOutputStream;

import org.apache.hadoop.fs.FileSystem;

import org.apache.hadoop.fs.Path;

import org.apache.hadoop.io.IOUtils;

public class UploadDataByStream {

    public static void main(String[] args) throws Exception {    

        Configuration conf = new Configuration();

        System.setProperty("HADOOP_USER_NAME", "admin");

        conf.set("fs.defaultFS", "hdfs://node21:9000");

        FileSystem fs = FileSystem.get(conf);

        InputStream in = new FileInputStream(new File("d:/abc.tar.gz"));

        FSDataOutputStream out = fs.create(new Path("/aa/abc.tar.gz"));

        IOUtils.copyBytes(in, out, 4096, true);

        fs.close();

    }

}

5、下载文件

package com.xyg.hdfs;

import java.io.File;

import java.io.FileOutputStream;

import java.io.OutputStream;

import org.apache.hadoop.conf.Configuration;

import org.apache.hadoop.fs.FSDataInputStream;

import org.apache.hadoop.fs.FileSystem;

import org.apache.hadoop.fs.Path;

import org.apache.hadoop.io.IOUtils;

public class DownloadDataByStream {

    public static void main(String[] args) throws Exception {

        Configuration conf = new Configuration();

        System.setProperty("HADOOP_USER_NAME", "admin");

        conf.set("fs.defaultFS", "hdfs://node21:9000");

        FileSystem fs = FileSystem.get(conf);

        FSDataInputStream in = fs.open(new Path("/aa/abc.tar.gz"));

        OutputStream out = new FileOutputStream(new File("D:/abc.sh"));

        IOUtils.copyBytes(in, out, 4096, true);

        fs.close();

    }

}

6、删除某个路径下特定类型的文件，比如class类型文件，比如txt类型文件

package com.xyg.hdfs;

import java.net.URI;

import org.apache.hadoop.conf.Configuration;

import org.apache.hadoop.fs.FileStatus;

import org.apache.hadoop.fs.FileSystem;

import org.apache.hadoop.fs.Path;

public class HDFS_DELETE_CLASS {

    public static final String FILETYPE = "tar.gz";

    public static final String DELETE_PATH = "/aa";

    public static void main(String[] args) throws Exception {

        new HDFS_DELETE_CLASS().rmrClassFile(new Path(DELETE_PATH));

    }

    public void rmrClassFile(Path path) throws Exception{

        // 首先获取集群必要的信息，以得到FileSystem的示例对象fs

        Configuration conf = new Configuration();

        FileSystem fs = FileSystem.get(new URI("hdfs://node2:9000"), conf, "admin");

        // 首先检查path本身是文件夹还是目录

        FileStatus fileStatus = fs.getFileStatus(path);

        boolean directory = fileStatus.isDirectory();

        // 根据该目录是否是文件或者文件夹进行相应的操作

        if(directory){

            // 如果是目录

            checkAndDeleteDirectory(path, fs);

        }else{

            // 如果是文件，检查该文件名是不是FILETYPE类型的文件

            checkAndDeleteFile(path, fs);

        }

    }

    // 处理目录

    public static void checkAndDeleteDirectory(Path path, FileSystem fs) throws Exception{

        // 查看该path目录下一级子目录和子文件的状态

        FileStatus[] listStatus = fs.listStatus(path);

        for(FileStatus fStatus: listStatus){

            Path p = fStatus.getPath();

            // 如果是文件，并且是以FILETYPE结尾，则删掉，否则继续遍历下一级目录

            if(fStatus.isFile()){

                checkAndDeleteFile(p, fs);

            }else{

                checkAndDeleteDirectory(p, fs);

            }

        }

    }

    // 檢查文件是否符合刪除要求，如果符合要求則刪除，不符合要求则不做处理

    public static void checkAndDeleteFile(Path path, FileSystem fs) throws Exception{

        String name = path.getName();

        System.out.println(name);

        /*// 直接判断有没有FILETYPE这个字符串,不是特别稳妥，并且会有误操作，所以得判断是不是以FILETYPE结尾

        if(name.indexOf(FILETYPE) != -1){

            fs.delete(path, true);

        }*/

        // 判断是不是以FILETYPE结尾

        int startIndex = name.length() - FILETYPE.length();

        int endIndex = name.length();

        // 求得文件后缀名

        String fileSuffix = name.substring(startIndex, endIndex);

        if(fileSuffix.equals(FILETYPE)){

            fs.delete(path, true);

        }

    }

}

7、删除HDFS集群中的所有空文件和空目录

public class DeleteEmptyDirAndFile {

    static FileSystem fs = null;

    public static void main(String[] args) throws Exception {

        initFileSystem();

//         创建测试数据

//        makeTestData();

        // 删除测试数据

//        deleteTestData();

        // 删除指定文件夹下的空文件和空文件夹

        deleteEmptyDirAndFile(new Path("/aa"));

    }

    /**

     * 删除指定文件夹下的 空文件 和 空文件夹

     * @throws Exception

     */

    public static void deleteEmptyDirAndFile(Path path) throws Exception {

        //当是空文件夹时

        FileStatus[] listStatus = fs.listStatus(path);

        if(listStatus.length == 0){

            fs.delete(path, true);

            return;

        }

        // 该方法的结果：包括指定目录的  文件 和 文件夹

        RemoteIterator<LocatedFileStatus> listLocatedStatus = fs.listLocatedStatus(path);

        while (listLocatedStatus.hasNext()) {

            LocatedFileStatus next = listLocatedStatus.next();

            Path currentPath = next.getPath();

            // 获取父目录

            Path parent = next.getPath().getParent();

            // 如果是文件夹，继续往下遍历，删除符合条件的文件（空文件夹）

            if (next.isDirectory()) {

                // 如果是空文件夹

                if(fs.listStatus(currentPath).length == 0){

                    // 删除掉

                    fs.delete(currentPath, true);

                }else{

                    // 不是空文件夹，那么则继续遍历

                    if(fs.exists(currentPath)){

                        deleteEmptyDirAndFile(currentPath);

                    }

                }

            // 如果是文件

            } else {

                // 获取文件的长度

                long fileLength = next.getLen();

                // 当文件是空文件时， 删除

                if(fileLength == 0){

                    fs.delete(currentPath, true);

                }

            }

            // 当空文件夹或者空文件删除时，有可能导致父文件夹为空文件夹，

            // 所以每次删除一个空文件或者空文件的时候都需要判断一下，如果真是如此，那么就需要把该文件夹也删除掉

            int length = fs.listStatus(parent).length;

            if(length == 0){

                fs.delete(parent, true);

            }

        }

    }

    /**

     * 初始化FileSystem对象之用

     */

    public static void initFileSystem() throws Exception{

        Configuration conf = new Configuration();

        System.setProperty("HADOOP_USER_NAME", "admin");

        conf.addResource("config/core-site.xml");

        conf.addResource("config/hdfs-site.xml");

        fs = FileSystem.get(conf);

    }

    /**

     * 创建 测试 数据之用

     */

    public static void makeTestData() throws Exception {

        String emptyFilePath = "D:\\bigdata\\1704mr_test\\empty.txt";

        String notEmptyFilePath = "D:\\bigdata\\1704mr_test\\notEmpty.txt";

        // 空文件夹 和 空文件 的目录

        String path1 = "/aa/bb1/cc1/dd1/";

        fs.mkdirs(new Path(path1));

        fs.mkdirs(new Path("/aa/bb1/cc1/dd2/"));

        fs.copyFromLocalFile(new Path(emptyFilePath), new Path(path1));

        fs.copyFromLocalFile(new Path(notEmptyFilePath), new Path(path1));

        // 空文件 的目录

        String path2 = "/aa/bb1/cc2/dd2/";

        fs.mkdirs(new Path(path2));

        fs.copyFromLocalFile(new Path(emptyFilePath), new Path(path2));

        // 非空文件 的目录

        String path3 = "/aa/bb2/cc3/dd3";

        fs.mkdirs(new Path(path3));

        fs.copyFromLocalFile(new Path(notEmptyFilePath), new Path(path3));

        // 空 文件夹

        String path4 = "/aa/bb2/cc4/dd4";

        fs.mkdirs(new Path(path4));

        System.out.println("测试数据创建成功");

    }

    /**

     * 删除 指定文件夹

     * @throws Exception

     */

    public static void deleteTestData() throws Exception {

        boolean delete = fs.delete(new Path("/aa"), true);

        System.out.println(delete ? "删除数据成功" : "删除数据失败");

    }

}

8、手动拷贝某个特定的数据块（比如某个文件的第二个数据块）

/**

     * 手动拷贝某个特定的数据块（比如某个文件的第二个数据块）

     * */

    public static void copyBlock(String str,int num) {

        Path path = new Path(str);

        BlockLocation[] localtions = new BlockLocation[0] ;

        try {

            FileStatus fileStatus = fs.getFileStatus(path);

            localtions = fs.getFileBlockLocations(fileStatus, 0, fileStatus.getLen());

            /*for(int i=0;i<localtions.length;i++) {

                System.out.println(localtions[i]);

            }*/

            /*System.out.println(localtions[num-1].getOffset());

            System.out.println(localtions[num-1].getLength());

            String[] hosts = localtions[num-1].getHosts();*/

            FSDataInputStream open = fs.open(path);

            open.seek(localtions[num-1].getOffset());

            OutputStream out = new FileOutputStream(new File("D:/abc.tar.gz"));

            IOUtils.copyBytes(open, out,4096,true);

        } catch (IOException e) {

            e.printStackTrace();

        }

    }

9、编写程序统计出HDFS文件系统中文件大小小于HDFS集群中的默认块大小的文件占比

import org.apache.hadoop.conf.Configuration;

import org.apache.hadoop.fs.FileSystem;

import org.apache.hadoop.fs.LocatedFileStatus;

import org.apache.hadoop.fs.Path;

import org.apache.hadoop.fs.RemoteIterator;

/**

 * 编写程序统计出HDFS文件系统中文件大小小于HDFS集群中的默认块大小的文件占比

 * 比如：大于等于128M的文件个数为98，小于128M的文件总数为2，所以答案是2%

 */

public class Exam1_SmallFilePercent {

    private static int DEFAULT_BLOCKSIZE = 128 * 1024 * 1024;

    public static void main(String[] args) throws Exception {

        Configuration conf = new Configuration();

        conf.set("fs.defaultFS", "hdfs://node21:9000");

        System.setProperty("HADOOP_USER_NAME", "admin");

        FileSystem fs = FileSystem.get(conf);

        Path path = new Path("/");

        float smallFilePercent = getSmallFilePercent(fs, path);

        System.out.println(smallFilePercent);

        fs.close();

    }

    /**

     * 该方法求出指定目录下的小文件和总文件数的对比

     * @throws Exception

     */

    private static float getSmallFilePercent(FileSystem fs, Path path) throws Exception {

        // TODO Auto-generated method stub

        int smallFile = 0;

        int totalFile = 0;

        RemoteIterator<LocatedFileStatus> listFiles = fs.listFiles(path, false);

        while(listFiles.hasNext()){

            totalFile++;

            LocatedFileStatus next = listFiles.next();

            long len = next.getLen();

            if(len < DEFAULT_BLOCKSIZE){

                smallFile++;

            }

        }

        System.out.println(smallFile+" : "+totalFile);

        return smallFile * 1f /totalFile;

    }

}

10、编写程序统计出HDFS文件系统中的平均数据块数（数据块总数/文件总数）

import org.apache.hadoop.conf.Configuration;

import org.apache.hadoop.fs.FileSystem;

import org.apache.hadoop.fs.LocatedFileStatus;

import org.apache.hadoop.fs.Path;

import org.apache.hadoop.fs.RemoteIterator;

/**

 * 编写程序统计出HDFS文件系统中的平均数据块数（数据块总数/文件总数）

 * 比如：一个文件有5个块，一个文件有3个块，那么平均数据块数为4

 * 如果还有一个文件，并且数据块就1个，那么整个HDFS的平均数据块数就是3

 */

public class Exam2_HDSFAvgBlocks {

    public static void main(String[] args) throws Exception {

        Configuration conf = new Configuration();

        conf.set("fs.defaultFS", "hdfs://node21:9000");

        System.setProperty("HADOOP_USER_NAME", "admin");

        FileSystem fs = FileSystem.get(conf);

        Path path = new Path("/");

        float avgHDFSBlocks = getHDFSAvgBlocks(fs, path);

        System.out.println("HDFS的平均数据块个数为：" + avgHDFSBlocks);

        fs.close();

    }

    /**

     * 求出指定目录下的所有文件的平均数据块个数

     */

    private static float getHDFSAvgBlocks(FileSystem fs, Path path) throws Exception {

        // TODO Auto-generated method stub

        int totalFiles = 0;        // 总文件数

        int totalBlocks = 0;    // 总数据块数

        RemoteIterator<LocatedFileStatus> listFiles = fs.listFiles(path, false);

        while(listFiles.hasNext()){

            LocatedFileStatus next = listFiles.next();

            int length = next.getBlockLocations().length;

            totalBlocks += length;

            if(next.getLen() != 0){

                totalFiles++;

            }

        }

        System.out.println(totalBlocks+" : "+totalFiles);

        return totalBlocks * 1f / totalFiles;

    }

}

11、编写程序统计出HDFS文件系统中的平均副本数（副本总数/总数据块数）

import org.apache.hadoop.conf.Configuration;

import org.apache.hadoop.fs.FileSystem;

import org.apache.hadoop.fs.LocatedFileStatus;

import org.apache.hadoop.fs.Path;

import org.apache.hadoop.fs.RemoteIterator;

/**

 * 编写程序统计出HDFS文件系统中的平均副本数（副本总数/总数据块数）

 * 比如：总共两个文件，一个文件5个数据块，每个数据块3个副本，第二个文件2个数据块，每个文件2个副本，最终的平均副本数 = （3*3 + 2*2）/（3+2）= 2.8

 */

public class Exam3_HDSFAvgBlockCopys {

    public static void main(String[] args) throws Exception {     

        Configuration conf = new Configuration();

        conf.set("fs.defaultFS", "hdfs://node21:9000");

        System.setProperty("HADOOP_USER_NAME", "admin");

        FileSystem fs = FileSystem.get(conf);

        Path path = new Path("/");

        float avgHDFSBlockCopys = getHDFSAvgBlockCopys(fs, path);

        System.out.println("HDFS的平均数据块个数为：" + avgHDFSBlockCopys);

        fs.close();

    }

    /**

     * 求出指定目录下的所有文件的平均数据块个数

     */

    private static float getHDFSAvgBlockCopys(FileSystem fs, Path path) throws Exception {

        // TODO Auto-generated method stub

        int totalCopy = 0;        // 总副本数

        int totalBlocks = 0;    // 总数据块数

        RemoteIterator<LocatedFileStatus> listFiles = fs.listFiles(path, false);

        while(listFiles.hasNext()){

            LocatedFileStatus next = listFiles.next();

            int length = next.getBlockLocations().length;

            short replication = next.getReplication();

            totalBlocks += length;

            totalCopy += length * replication;

        }

        System.out.println(totalCopy+" : "+totalBlocks);

        return totalCopy * 1f / totalBlocks;

    }

}

12、统计HDFS整个文件系统中的不足指定数据块大小的数据块的比例

import java.io.IOException;

import org.apache.hadoop.conf.Configuration;

import org.apache.hadoop.fs.BlockLocation;

import org.apache.hadoop.fs.FileSystem;

import org.apache.hadoop.fs.LocatedFileStatus;

import org.apache.hadoop.fs.Path;

import org.apache.hadoop.fs.RemoteIterator;

/**

 * 统计HDFS整个文件系统中的不足指定数据块大小的数据块的比例

 * 比如指定的数据块大小是128M，总数据块有100个，不是大小为完整的128M的数据块有5个，那么不足指定数据块大小的数据块的比例就为5%

 * 注意：千万注意考虑不同文件的指定数据块大小可能不一致。所以千万不能用默认的128M一概而论

 */

public class Exam4_LTBlockSize {

    public static void main(String[] args) throws Exception {

        Configuration conf = new Configuration();

        conf.set("fs.defaultFS", "hdfs://node21:9000");

        System.setProperty("HADOOP_USER_NAME", "admin");

        FileSystem fs = FileSystem.get(conf);

        Path path = new Path("/");

        float avgHDFSBlockCopys = getLessThanBlocksizeBlocks(fs, path);

        System.out.println("HDFS的不足指定数据块大小的数据块数目为：" + avgHDFSBlockCopys);

        fs.close();

    }

    private static float getLessThanBlocksizeBlocks(FileSystem fs, Path path) throws Exception {

        // TODO Auto-generated method stub

        int totalBlocks = 0;                // 总副本数

        int lessThenBlocksizeBlocks = 0;    // 总数据块数

        RemoteIterator<LocatedFileStatus> listFiles = fs.listFiles(path, false);

        while(listFiles.hasNext()){

            LocatedFileStatus next = listFiles.next();

            BlockLocation[] blockLocations = next.getBlockLocations();

            int length = blockLocations.length;

            if(length != 0){

                totalBlocks += length;

                long lastBlockSize = blockLocations[length - 1].getLength();

                long blockSize = next.getBlockSize();

                if(lastBlockSize < blockSize){

                    lessThenBlocksizeBlocks++;

                }

            }

        }

        System.out.println(lessThenBlocksizeBlocks+" : "+totalBlocks);

        return lessThenBlocksizeBlocks * 1f / totalBlocks;

    }

}

13、统计出一个给定数组的蓄水总量（把数组的每个位置的数看是做地势高低）

/**

        统计出一个给定数组的蓄水总量（把数组的每个位置的数看是做地势高低）

        比如：int[] intArray = new int[]{4,3,2,5,6,4,4,7}

        能蓄水：[0,1,2,0,0,2,2,0] 所以总量是：7

    核心思路：把数组切成很多个 01数组，每一层一个01数组，统计每个01数组中的合法0的总个数（数组的左边第一个1的中间区间中的0的个数）即可

 */

public class Exam5_WaterStoreOfArray {

    public static void main(String[] args) {

//        int[] intArray = new int[]{4,3,2,5,6,4,4,7};

//        int[] intArray = new int[]{1,2,3,4,5,6};

        int[] intArray = new int[]{3,1,2,7,3,8,4,9,5,6};

        int totalWater = getArrayWater(intArray);

        System.out.println(totalWater);

    }

    /**

     * 求出数组中的水数

     */

    private static int getArrayWater(int[] intArray) {

        int findMaxValueOfArray = findMaxValueOfArray(intArray);

        int findMinValueOfArray = findMinValueOfArray(intArray);

        int length = intArray.length;

        int totalWater = 0;

        // 循环次数就是最大值和最小值的差

        for(int i=findMinValueOfArray; i<findMaxValueOfArray; i++){

            // 循环构造每一层的01数组

            int[] tempArray = new int[length];

            for(int j=0; j<length; j++){

                if(intArray[j] > i){

                    tempArray[j] = 1;

                }else{

                    tempArray[j] = 0;

                }

            }

            // 获取每一个01数组的合法0个数

            int waterOfOneZeroArray = getWaterOfOneZeroArray(tempArray);

            totalWater += waterOfOneZeroArray;

        }

        return totalWater;

    }

    /**

     * 寻找逻辑是：从左右开始各找一个1，然后这两个1之间的所有0的个数，就是水数

     */

    private static int getWaterOfOneZeroArray(int[] tempArray) {

        int length = tempArray.length;

        int toatalWater = 0;

        // 找左边的1

        int i = 0;

        while(i < length){

            if(tempArray[i] == 1){

                break;

            }

            i++;

        }

        // 从右边开始找1

        int j=length-1;

        while(j >= i){

            if(tempArray[j] == 1){

                break;

            }

            j--;

        }

        // 找以上两个1之间的0的个数。

        if(i == j || i + 1 == j){

            return 0;

        }else{

            for(int k=i+1; k<j; k++){

                if(tempArray[k] == 0){

                    toatalWater++;

                }

            }

            return toatalWater;

        }

    }

    /**

     *

     * 描述：找出一个数组中的最大值

     */

    public static int findMaxValueOfArray(int[] intArray){

        int length = intArray.length;

        if(length == 0){

            return 0;

        }else if(length == 1){

            return intArray[0];

        }else{

            int max = intArray[0];

            for(int i=1; i<length; i++){

                if(intArray[i] > max){

                    max = intArray[i];

                }

            }

            return max;

        }

    }

    /**

     * 找出一个数组中的最小值

     */

    public static int findMinValueOfArray(int[] intArray){

        int length = intArray.length;

        if(length == 0){

            return 0;

        }else if(length == 1){

            return intArray[0];

        }else{

            int min = intArray[0];

            for(int i=1; i<length; i++){

                if(intArray[i] < min){

                    min = intArray[i];

                }

            }

            return min;

        }

    }

}

巴特西

Hadoop（四）HDFS的高级API操作

一 HDFS客户端环境准备

1.1 jar包准备

1.2 Eclipse准备

二 HDFS的高级API编程

2.1 HDFS文件上传（测试参数优先级）

参数优先级：（1）客户端代码中设置的值 >（2）classpath下的用户自定义配置文件 >（3）然后是服务器的默认配置

2.2 文件的上传和下载

最新文章

热门文章

巴特西

Hadoop（四）HDFS的高级API操作

一 HDFS客户端环境准备

1.1 jar包准备

1.2 Eclipse准备

二 HDFS的高级API编程

2.1 HDFS文件上传（测试参数优先级）

参数优先级： （1）客户端代码中设置的值 >（2）classpath下的用户自定义配置文件 >（3）然后是服务器的默认配置

2.2 文件的上传和下载

最新文章

热门文章

参数优先级：（1）客户端代码中设置的值 >（2）classpath下的用户自定义配置文件 >（3）然后是服务器的默认配置