Hadoop-JAVA编写HDFS客户端进行HDFS操作

野性酷女 2022-10-11 03:56 159阅读 0赞

## HDFS客户端操作 ##

### 环境准备 ###

根据电脑操作系统将对应的编译后的jar包放到非中文路径下，我的是Linux因此将Linux编译后的jar放到指定目录下`/home/lxj/workspace/hadoop-2.7.0`

配置环境变量，然后使其生效

export HADOOP_HOME=/home/lxj/workspace/hadoop-2.7.2
    export PATH=$PATH:$HADOOP_HOME/bin
    export PATH=$PATH:$HADOOP_HOME/sbin

然后创建一个简单的maven项目，jdk使用1.8，将以下依赖导入：

<dependencies>
        <dependency>
            <groupId>junit</groupId>
            <artifactId>junit</artifactId>
            <version>RELEASE</version>
        </dependency>
            <dependency>
            <groupId>org.apache.logging.log4j</groupId>
            <artifactId>log4j-core</artifactId>
            <version>2.8.2</version>
        </dependency>
        <dependency>
            <groupId>org.apache.hadoop</groupId>
            <artifactId>hadoop-common</artifactId>
            <version>2.7.2</version>
        </dependency>
        <dependency>
            <groupId>org.apache.hadoop</groupId>
            <artifactId>hadoop-client</artifactId>
            <version>2.7.2</version>
        </dependency>
        <dependency>
            <groupId>org.apache.hadoop</groupId>
            <artifactId>hadoop-hdfs</artifactId>
            <version>2.7.2</version>
        </dependency>
        <dependency>
            <groupId>jdk.tools</groupId>
            <artifactId>jdk.tools</artifactId><version>1.8</version>
            <scope>system</scope>
            <systemPath>${JAVA_HOME}/lib/tools.jar</systemPath>
        </dependency>
    </dependencies>

### 环境测试 ###

创建一个main方法

public class HDFSClient { 
    
        public static void main (String[] args) throws IOException { 
    
            // 1. 获取hdfs客户端对象
            
            // 方法一：此方法需要用到以下的指定集群用户
            Configuration conf = new Configuration();
            // 设置要访问的集群，值可以从已有集群的core-site.xml文件获取，或者直接指定
            // <property>
            // <name>fs.defaultFS</name>
            // <value>hdfs://hadoop113:9000</value>
            // </property>
            // 同时本地要配置对应的域名映射
            conf.set("fs.defaultFS", "hdfs://hadoop113:9000");
            FileSystem fs = FileSystem.get(conf);
    
            // 方法二，一步到位
            FileSystem fs = FileSystem.get(new URI("hdfs://hadoop113:9000"), conf, "bd");
            
            
            // 2.hdfs上创建路径
            fs.mkdirs(new Path("/sanguo/shu"));
    
            // 3.关闭资源
            fs.close();
    
            System.out.println("over");
        }
    }

指定集群对应的用户bd。

IDEA中指定集群用户如下（-DHADOOP\_USER\_NAME=bd）：

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2NsZWFybHhq_size_16_color_FFFFFF_t_70]

然后运行程序，从网页上可以看到/sanguo/shu目录已经添加成功了  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2NsZWFybHhq_size_16_color_FFFFFF_t_70 1]

### HDFS参数优先级 ###

HDFS的参数优先级的话，如下：

代码中的`Configuration conf = new Configuration()`的set方法优先级最高；

src/main/resources目录下的xml文件优先级第二；

集群上的xml优先级最低。

1、例如默认的dfs.replication是3，即默认副本数是3，如果在resources目录下有hdfs-site.xml如下配置文件，则副本数为1

<?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>

2、默认为3，且resources下的文件配置为1，如果在代码中显示配置了副本数的为2的话，那么最终上传之后的副本数为2

Configuration configuration = new Configuration();
    configuration.set("dfs.replication", "2");

### HDFS常用API的使用 ###

public class HDFSClient { 
    
        public final static String HDFS_URI = "hdfs://hadoop113:9000";
        public final static String HDFS_USER = "bd";
        public final static String FILE_PRE = "/home/lxj/hadoop-data/";
        public final static String HDFS_FILE_PRE = "/sanguo/shu/";
    
        public static void main (String[] args) throws Exception { 
    
            // 获取hdfs客户端对象
            Configuration conf = new Configuration();
            FileSystem fs = FileSystem.get(new URI(HDFS_URI), conf, HDFS_USER);
    
            // mkdirs
            // 创建目录
             fs.mkdirs(new Path("/sanguo/shu"));
    
            // copyFromLocalFile
            // 上传文件
             fs.copyFromLocalFile(getLocalPathByFile("guanyu"), getHDFSPathByFile("guanyu"));
    
            // copyToLocalFile
            // 下载文件
            // 1、这样下载的话（非本地模式），会多产生一个crc文件，可以用来校验文件的完整性以及正确性
             fs.copyToLocalFile(getHDFSPathByFile("guanyu"), getLocalPathByFile("guanyu_copy"));
            // 2、这个方法有4个参数
            /** * delSrc: 默认为false，如果是true的话，将会删除源文件，移动文件实际调用的方法就是false情况的本方法 * src: 源文件路径 * dest: 目标文件路径 * useRowLocalFileSystem: 是否使用本地模式 */
             fs.copyToLocalFile(false, getHDFSPathByFile("guanyu"), getLocalPathByFile("guanyu_copy2"), true);
    
            // delete
            // 删除文件
            // 删除文件时，第二个参数任意
             fs.delete(getHDFSPathByFile("guanyu"), false);
            // 删除文件夹时，第二个参数必须为true，表示递归删除
             fs.mkdirs(new Path("/sanguo/han"));
             fs.delete(new Path("/sanguo/han"), true);
    
            // rename
            // 文件更名
             fs.copyFromLocalFile(getLocalPathByFile("guanyu"), getHDFSPathByFile("guanyu"));
             fs.rename(getHDFSPathByFile("guanyu"), getHDFSPathByFile("guanyunchang"));
            // 文件夹更名
             fs.mkdirs(new Path("/sanguo/han"));
             fs.rename(new Path("/sanguo/han"), new Path("/sanguo/donghan"));
    
            // listFiles
            // 文件详情查看
            RemoteIterator<LocatedFileStatus> listFiles = fs.listFiles(new Path("/"), true);
    
            while (listFiles.hasNext()) { 
                LocatedFileStatus fileStatus = listFiles.next();
    
                // 查看文件的名称、权限、长度、块信息等
                System.out.println(fileStatus.getPath().getName()); // 名称
                System.out.println(fileStatus.getPermission()); // 权限
                System.out.println(fileStatus.getLen()); // 长度
    
                BlockLocation[] blockLocations = fileStatus.getBlockLocations();
                for (BlockLocation blockLocation : blockLocations) { 
                    String[] hosts = blockLocation.getHosts();
                    for (String host : hosts) { 
                        System.out.println(host);
                    }
                }
    
                System.out.println("-----------分割线------------------------------");
            }
    
            // listStatus
            // 判断是文件还是文件夹
            FileStatus[] listStatus = fs.listStatus(new Path("/"));
            for (FileStatus fileStatus : listStatus) { 
                if (fileStatus.isFile()) { 
                    // 文件
                    System.out.println("file: " + fileStatus.getPath().getName());
                } else { 
                    // 文件夹
                    System.out.println("dir: " + fileStatus.getPath().getName());
                }
            }
    
    
            // 关闭资源
            fs.close();
        }
    
        private static Path getLocalPathByFile (String filename) { 
    
            return new Path(FILE_PRE + filename);
        }
    
        private static Path getHDFSPathByFile (String filename) { 
    
            return new Path(HDFS_FILE_PRE + filename);
        }
    }

### HDFS的I/O流操作 ###

#### 上传文件 ####

此部分代码HDFSClient下

// fs.create(Path) 获取HDFS的输出流，用于文件上传
    @Test
    public void putFileToHDFS() throws Exception { 
    
        // 获取hdfs客户端对象
        Configuration conf = new Configuration();
        FileSystem fs = FileSystem.get(new URI(HDFS_URI), conf, HDFS_USER);
    
        // 获取输入流
        FileInputStream fis = new FileInputStream(new File(FILE_PRE + "guanyu"));
    
        // 获取输出流
        FSDataOutputStream fos = fs.create(new Path("/ioguanyu"));
    
        // 流的对拷
        IOUtils.copyBytes(fis, fos, conf);
    
        // 关闭资源
        IOUtils.closeStream(fis);
        IOUtils.closeStream(fos);
        fs.close();
    }

#### 下载文件 ####

此部分代码HDFSClient下

// fs.open(Path) 获取HDFS的输入流，用于文件下载
    @Test
    public void getFileToLocal() throws Exception { 
    
        // 获取hdfs客户端对象
        Configuration conf = new Configuration();
        FileSystem fs = FileSystem.get(new URI(HDFS_URI), conf, HDFS_USER);
    
        // 获取输入流
        FSDataInputStream fis = fs.open(new Path("/ioguanyu"));
    
        // 获取输出流
        FileOutputStream fos = new FileOutputStream(new File(FILE_PRE + "guanyuio"));
    
        // 流的对拷
        IOUtils.copyBytes(fis, fos, conf);
    
        // 关闭资源
        IOUtils.closeStream(fis);
        IOUtils.closeStream(fos);
        fs.close();
    }

#### 定位文件读取 ####

作用是分块读取HDFS上的大文件，比如一个文件是200MB，而默认的块大小是128MB，那么该文件会被分为两块，一块128MB，另一块是72MB。可以只下载我们需要的块，比如日志文件很大，但是只需要下载最近的一些数据，那么就可以用到定位文件读取了。

// 读取第一块
    @Test
    public void readFileSeek1() throws Exception { 
    
        // 获取hdfs客户端对象
        Configuration conf = new Configuration();
        FileSystem fs = FileSystem.get(new URI(HDFS_URI), conf, HDFS_USER);
    
        // 获取输入流
        FSDataInputStream fis = fs.open(new Path("/hadoop-2.7.2.tar.gz"));
    
        // 获取输出流
        FileOutputStream fos = new FileOutputStream(new File(FILE_PRE + "hadoop-2.7.2.tar.gz.part1"));
    
        // 流的对拷（只拷贝128M）
        byte[] buf = new byte[1024];
        for (int i = 0; i < 1024 * 128; i++) { 
            fis.read(buf);
            fos.write(buf);
        }
    
        // 关闭资源
        IOUtils.closeStream(fis);
        IOUtils.closeStream(fos);
        fs.close();
    }
    
    // 读取第二块
    // 使用fis.seek(start)来设置读取的起点
    @Test
    public void readFileSeek2() throws Exception { 
    
        // 获取hdfs客户端对象
        Configuration conf = new Configuration();
        FileSystem fs = FileSystem.get(new URI(HDFS_URI), conf, HDFS_USER);
    
        // 获取输入流
        FSDataInputStream fis = fs.open(new Path("/hadoop-2.7.2.tar.gz"));
    
        // 获取输出流
        FileOutputStream fos = new FileOutputStream(new File(FILE_PRE + "hadoop-2.7.2.tar.gz.part2"));
    
        // 指定读取的起点，读取128M之后的文件
        fis.seek(1024 * 1024 * 128);
    
        // 流的对拷
        IOUtils.copyBytes(fis, fos, conf);
    
        // 关闭资源
        IOUtils.closeStream(fis);
        IOUtils.closeStream(fos);
        fs.close();
    }

两个文件下载成功之后，验证是否下载成功，那么使用cat命令将其拼接来验证。

cat hadoop-2.7.2.tar.gz.part1 hadoop-2.7.2.tar.gz.part2 > hadoop-2.7.2.tar.gz
    
    ## 拼接之后的文件能够正常使用，说明下载成功

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2NsZWFybHhq_size_16_color_FFFFFF_t_70]: /images/20221005/adaab18f801b4941816f758cf41d38ba.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2NsZWFybHhq_size_16_color_FFFFFF_t_70 1]: /images/20221005/b3e6b6cb85384e108b456d7f29485f62.png