HDFS的API操作

程序员文章站 2024-03-23 08:05:10

...

文章目录

创建maven工程并导入jar包
使用url的方式访问数据
使用文件系统方式访问数据
获取FileSystem的几种方式
递归遍历文件系统当中的所有文件
下载文件到本地
hdfs上面创建文件夹
文件的上传操作
读取文件
小文件的合并

创建maven工程并导入jar包

由于cdh版本的所有的软件涉及版权的问题，所以并没有将所有的jar包托管到maven仓库当中去，而是托管在了CDH自己的服务器上面，所以我们默认去maven的仓库下载不到，需要自己手动的添加repository去CDH仓库进行下载，以下两个地址是官方文档说明，请仔细查阅
https://www.cloudera.com/documentation/enterprise/release-notes/topics/cdh_vd_cdh5_maven_repo.html
https://www.cloudera.com/documentation/enterprise/release-notes/topics/cdh_vd_cdh5_maven_repo_514x.html

创建maven工程，pom配置
HDFS的API操作

<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0"
         xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
    <parent>
        <artifactId>bigdata13</artifactId>
        <groupId>cn.itcast.bigdata13</groupId>
        <version>1.0-SNAPSHOT</version>
    </parent>
    <modelVersion>4.0.0</modelVersion>

    <artifactId>day03_hdfs</artifactId>

    <repositories>
        <repository>
            <id>cloudera</id>
            <url>https://repository.cloudera.com/artifactory/cloudera-repos/</url>
        </repository>
    </repositories>


    <dependencies>
        <dependency>
            <groupId>org.apache.hadoop</groupId>
            <artifactId>hadoop-client</artifactId>
            <version>2.6.0-mr1-cdh5.14.0</version>
        </dependency>
        <dependency>
            <groupId>org.apache.hadoop</groupId>
            <artifactId>hadoop-common</artifactId>
            <version>2.6.0-cdh5.14.0</version>
        </dependency>
        <dependency>
            <groupId>org.apache.hadoop</groupId>
            <artifactId>hadoop-hdfs</artifactId>
            <version>2.6.0-cdh5.14.0</version>
        </dependency>

        <dependency>
            <groupId>org.apache.hadoop</groupId>
            <artifactId>hadoop-mapreduce-client-core</artifactId>
            <version>2.6.0-cdh5.14.0</version>
        </dependency>
        <!-- https://mvnrepository.com/artifact/junit/junit -->
        <dependency>
            <groupId>junit</groupId>
            <artifactId>junit</artifactId>
            <version>4.11</version>
            <scope>test</scope>
        </dependency>
        <dependency>
            <groupId>org.testng</groupId>
            <artifactId>testng</artifactId>
            <version>RELEASE</version>
            <scope>test</scope>
        </dependency>
    </dependencies>
    <build>
        <plugins>
            <plugin>
                <groupId>org.apache.maven.plugins</groupId>
                <artifactId>maven-compiler-plugin</artifactId>
                <version>3.0</version>
                <configuration>
                    <source>1.8</source>
                    <target>1.8</target>
                    <encoding>UTF-8</encoding>
                    <!--    <verbal>true</verbal>-->
                </configuration>
            </plugin>


            <!--  打包的插件，我们项目当中需要用到的其他的jar包都打到一起去
              -->
            <plugin>
                <groupId>org.apache.maven.plugins</groupId>
                <artifactId>maven-shade-plugin</artifactId>
                <version>2.4.3</version>
                <executions>
                    <execution>
                        <phase>package</phase>
                        <goals>
                            <goal>shade</goal>
                        </goals>
                        <configuration>
                            <minimizeJar>true</minimizeJar>
                        </configuration>
                    </execution>
                </executions>
            </plugin>
            <!--  <plugin>
                  <artifactId>maven-assembly-plugin </artifactId>
                  <configuration>
                      <descriptorRefs>
                          <descriptorRef>jar-with-dependencies</descriptorRef>
                      </descriptorRefs>
                      <archive>
                          <manifest>
                              <mainClass>cn.itcast.hadoop.db.DBToHdfs2</mainClass>
                          </manifest>
                      </archive>
                  </configuration>
                  <executions>
                      <execution>
                          <id>make-assembly</id>
                          <phase>package</phase>
                          <goals>
                              <goal>single</goal>
                          </goals>
                      </execution>
                  </executions>
              </plugin>-->
        </plugins>
    </build>







</project>

使用url的方式访问数据

import org.apache.commons.io.IOUtils;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.*;
import org.junit.Test;
import sun.security.krb5.internal.PAData;

import java.io.*;
import java.net.MalformedURLException;
import java.net.URI;
import java.net.URISyntaxException;
import java.net.URL;

public class HdfsOperate {
    @Test
    public  void  downHdfsFile() throws Exception {

        //第一步：注册hdfs的驱动文件，表示我们使用hdfs://这种协议访问我们的hdfs文件系统
        URL.setURLStreamHandlerFactory(new FsUrlStreamHandlerFactory());
        //URL需要接受一个文件访问的地址
        String  url = "hdfs://node01:8020/install2.log";
        //获取到了一个inputStream
        InputStream inputStream = new URL(url).openStream();

        OutputStream outputStream = new FileOutputStream(new File("c:\\hello.txt"));

        //通过工具类，将我们 的输入流，读取到输出流，里面去
        IOUtils.copy(inputStream,outputStream);
        //关闭io流
        IOUtils.closeQuietly(inputStream);
        IOUtils.closeQuietly(outputStream);

    }

如果执行出现以下错误，可以参见资料如何解决，也可以不用理会，不会影响程序的执行。记得配置完成环境变量之后重启开发工具
HDFS的API操作

使用文件系统方式访问数据

在 java 中操作 HDFS，主要涉及以下 Class：

Configuration：该类的对象封装了客户端或者服务器的配置;
FileSystem：该类的对象是一个文件系统对象，可以用该对象的一些方法来对文件进行操作，通过 FileSystem 的静态方法 get 获得该对象。
FileSystem fs = FileSystem.get(conf)
get 方法从 conf 中的一个参数 fs.defaultFS 的配置值判断具体是什么类型的文件系统。如果我们的代码中没有指定 fs.defaultFS，并且工程 classpath下也没有给定相应的配置，conf中的默认值就来自于hadoop的jar包中的core-default.xml ，默认值为： file:/// ，则获取的将不是一个DistributedFileSystem 的实例，而是一个本地文件系统的客户端对象

获取FileSystem的几种方式

   /**
     * 获取文件系统的几种方式
     */
    @Test
    public  void getFileSystem1() throws IOException {
        //FileSystem是一个抽象类，获取抽象类的实例 有两种方式，第一种，看看这个抽象类有么有提供什么方法，返回他本身
        //第二种方式，找子类
        Configuration configuration = new Configuration();
        //如果这里不加任何配置，这里获取到的就是本地文件系统，现在用node01覆盖它
        configuration.set("fs.defaultFS","hdfs://node01:8020");
        FileSystem fileSystem = FileSystem.get(configuration);
        System.out.println(fileSystem.toString());
        fileSystem.close();

    }

    /**
     * 第二种方式获取hdfs分布式文件系统
     */
    @Test
    public  void  getFileSystem2() throws Exception {
        Configuration configuration = new Configuration();
        //通过我们指定URI来获取分布式文件系统
        FileSystem fileSystem = FileSystem.get(new URI("hdfs://node01:8020"), configuration);
        System.out.println(fileSystem.toString());
        fileSystem.close();
    }


    /**
     * 第三种获取分布式文件系统的方法
     */
    @Test
    public  void getFileSystem3() throws IOException {
        Configuration configuration = new Configuration();
        configuration.set("fs.defaultFS","hdfs://node01:8020");
        FileSystem fileSystem = FileSystem.newInstance(configuration);
        System.out.println(fileSystem.toString());
        fileSystem.close();
    }

递归遍历文件系统当中的所有文件

    @Test
    public  void  getAllFiles()throws  Exception{
        //获取文件系统
        FileSystem fileSystem = FileSystem.get(new URI("hdfs://node01:8020"), new Configuration());

        //获取到我们的文件的状态，可以通过fileStatuses来判断究竟是文件还是文件夹
        FileStatus[] fileStatuses = fileSystem.listStatus(new Path("hdfs://node01:8020/"));
        //循环遍历FileStatus  判断文件究竟是文件夹，还是文件，如果是文件，直接输出路径，如果是文件夹，继续进去遍历
        for (FileStatus fileStatus : fileStatuses) {
            if(fileStatus.isDirectory()){
                //文件夹
            getDirFiles(fileStatus.getPath(),fileSystem);

            }else{
                //文件
                Path path = fileStatus.getPath();
                System.out.println(path.toString());

            }

        }



    }

    public  void  getDirFiles(Path path,FileSystem fileSystem) throws IOException {
        FileStatus[] fileStatuses = fileSystem.listStatus(path);
        for (FileStatus fileStatus : fileStatuses) {
            if(fileStatus.isDirectory()){
                getDirFiles(fileStatus.getPath(),fileSystem);
            }else{
                System.out.println( fileStatus.getPath().toString());


            }
        }



    }

API方法：

 /***
     * 通过hdfs直接提供的api进行遍历
     */
    @Test
    public void  listALLFiles2() throws  Exception{
        //获取文件系统
        FileSystem fileSystem = FileSystem.get(new URI("hdfs://node01:8020"), new Configuration());
        RemoteIterator<LocatedFileStatus> locatedFileStatusRemoteIterator = fileSystem.listFiles(new Path("hdfs://node01:8020/"), true);
        while (locatedFileStatusRemoteIterator.hasNext()){
            LocatedFileStatus fileStatus = locatedFileStatusRemoteIterator.next();
            System.out.println(fileStatus.getPath().toString());

        }
        fileSystem.close();
    }

下载文件到本地

   /**
     * 下载hdfs的文件到本地目录
     */
    @Test
    public void downFileTolocal() throws  Exception{
        //获取文件系统
        FileSystem fileSystem = FileSystem.get(new URI("hdfs://node01:8020"), new Configuration());
      //第一种下载文件方式
       /* FSDataInputStream inputStream = fileSystem.open(new Path("hdfs://node01:8020/install2.log"));
        FileOutputStream fileOutputStream = new FileOutputStream(new File("c:\\hello2.txt"));
        IOUtils.copy(inputStream,fileOutputStream);
        IOUtils.closeQuietly(inputStream);
        IOUtils.closeQuietly(fileOutputStream);
        fileSystem.close();*/

       //第二种文件下载的方式
        fileSystem.copyToLocalFile(new Path("hdfs://node01:8020/install2.log"),new Path("file:///c:\\hello3.txt"));
        fileSystem.close();

    }

hdfs上面创建文件夹

/**
 * hdfs上面创建文件夹
 */
@Test
public   void createDir() throws Exception {
    FileSystem fileSystem = FileSystem.newInstance(new URI("hdfs://node01:8020"), new Configuration());
    Path path = new Path("hdfs://node01:8020/dir1/dir2");
    fileSystem.mkdirs(path);
    //关闭客户端
    fileSystem.close();
}

文件的上传操作

    /**
     * 文件的上传操作
     */
    @Test
    public void uploadFile() throws  Exception{
        FileSystem fileSystem = FileSystem.newInstance(new URI("hdfs://node01:8020"), new Configuration());
        fileSystem.copyFromLocalFile(new Path("file:///c:\\hello3.txt"),new Path("hdfs://node01:8020/dir1/dir2/hello.txt"));
        fileSystem.close();

//思考，如何通过io流的方式操作我们的文件上传

      /*  FSDataOutputStream fsDataOutputStream = fileSystem.create(new Path("hdfs://node01:8020/dir1/dir2/abc.txt"));
        FileInputStream fileInputStream = new FileInputStream(new File("c:\\hello3.txt"));
*/

    }

读取文件

/**
 * 读取文件
 */
    @Test
    public  void  readFile() throws  Exception{
        FileSystem fileSystem = FileSystem.newInstance(new URI("hdfs://node01:8020"), new Configuration(),"root");

        fileSystem.copyToLocalFile(new Path("hdfs://node01:8020/config/core-site.xml"),new Path("file:///c:\\core-site.xml"));

        fileSystem.close();


    }

小文件的合并

  /*
    小文件的合并
     */
    @Test
    public   void  mergeSmallFile() throws  Exception{
        //获取分布式文件系统
        FileSystem fileSystem = FileSystem.newInstance(new URI("hdfs://node01:8020"), new Configuration(),"root");
        //获取hdfs文件上面的输出流
        FSDataOutputStream fsDataOutputStream = fileSystem.create(new Path("hdfs://node01:8020/bigFile.xml"));

        //获取本地文件系统，遍历得到我们本地的小文件，将每一个小文件读成一个输入流
        LocalFileSystem localFileSystem = FileSystem.getLocal(new Configuration());
        //通过本地文件系统，遍历本地的所有的小文件
        FileStatus[] fileStatuses = localFileSystem.listStatus(new Path("file:///F:\\传智播客大数据离线阶段课程资料\\3、大数据离线第三天\\上传小文件合并"));
        for (FileStatus fileStatus : fileStatuses) {
            //获取每一个文件的路径
            Path path = fileStatus.getPath();
            //循环遍历读取我们每一个文件的输入流
            FSDataInputStream fsDataInputStream = localFileSystem.open(path);
            IOUtils.copy(fsDataInputStream, fsDataOutputStream);
            IOUtils.closeQuietly(fsDataInputStream);



        }

        IOUtils.closeQuietly(fsDataOutputStream);




        //第一步：获取本地文件系统

        //第二步：找到所有的本地文件系统的小文件


        //第三步：输出成IO流

        //第四步：将我们的IO流输出到hdfs上面去

相关标签：大数据 hadoop java maven hdfs

上一篇：阿里云虚拟机外网访问yarn 8088 hdfs 50070

下一篇：【hadoop】hadoop集群介绍和完全分布式部署步骤

HDFS的API操作

文章目录

创建maven工程并导入jar包

使用url的方式访问数据

使用文件系统方式访问数据

获取FileSystem的几种方式

递归遍历文件系统当中的所有文件

下载文件到本地

hdfs上面创建文件夹

文件的上传操作

读取文件

小文件的合并

HDFS相关操作

Hadoop入门之eclipse对HDFS的常用操作

HDFS java API操作

java api操作HDFS

HDFS Java api操作

Hadoop学习笔记(2)-HDFS的基本操作(Shell命令)

Hadoop 实验: 熟悉常用的HDFS操作

大数据_07 【hadoop HDFS的shell命令操作】

JAVA操作HDFS案例

使用Hadoop的Java API操作HDFS