大数据开发之HDFS的API操作过程

创建maven工程并导入jar包

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

        <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>
    </plugins>
</build>

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

在 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的几种方式

第一种方式获取FileSystem

@Test
public void getFileSystem() throws URISyntaxException, IOException {
   Configuration configuration = new Configuration();
    FileSystem fileSystem = FileSystem.get(new URI("hdfs://192.168.47.100:8020"), configuration);
    System.out.println(fileSystem.toString());
}

第二种方式获取FileSystem

@Test
public void getFileSystem2() throws URISyntaxException, IOException {
    Configuration configuration = new Configuration();
    configuration.set("fs.defaultFS","hdfs://192.168.47.100:8020");
    FileSystem fileSystem = FileSystem.get(new URI("/"), configuration);
    System.out.println(fileSystem.toString());
}

第三种获取FileSystem类的方式

@Test
public void getFileSystem3() throws URISyntaxException, IOException {
    Configuration configuration = new Configuration();
    FileSystem fileSystem = FileSystem.newInstance(new URI("hdfs://192.168.47.100:8020"), configuration);
    System.out.println(fileSystem.toString());
}

第四种获取FileSystem类的方式

@Test
public void getFileSystem4() throws  Exception{
    Configuration configuration = new Configuration();
    configuration.set("fs.defaultFS","hdfs://192.168.47.100:8020");
    FileSystem fileSystem = FileSystem.newInstance(configuration);
    System.out.println(fileSystem.toString());
}

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

通过递归遍历hdfs文件系统

@Test
public void listFile() throws Exception{
    FileSystem fileSystem = FileSystem.get(new URI("hdfs://192.168.47.100:8020"), new Configuration());
    FileStatus[] fileStatuses = fileSystem.listStatus(new Path("/"));
    for (FileStatus fileStatus : fileStatuses) {
        if(fileStatus.isDirectory()){
            Path path = fileStatus.getPath();
            listAllFiles(fileSystem,path);
        }else{
            System.out.println("文件路径为"+fileStatus.getPath().toString());
        }
    }
}
public void listAllFiles(FileSystem fileSystem,Path path) throws  Exception{
    FileStatus[] fileStatuses = fileSystem.listStatus(path);
    for (FileStatus fileStatus : fileStatuses) {
        if(fileStatus.isDirectory()){
            listAllFiles(fileSystem,fileStatus.getPath());
        }else{
            Path path1 = fileStatus.getPath();
            System.out.println("文件路径为"+path1);
        }
    }
}

官方提供的API直接遍历

/**
 * 递归遍历官方提供的API版本
 * @throws Exception
 */
@Test
public void listMyFiles()throws Exception{
    //获取fileSystem类
    FileSystem fileSystem = FileSystem.get(new URI("hdfs://192.168.52.100:8020"), new Configuration());
    //获取RemoteIterator 得到所有的文件或者文件夹,第一个参数指定遍历的路径,第二个参数表示是否要递归遍历
    RemoteIterator<LocatedFileStatus> locatedFileStatusRemoteIterator = fileSystem.listFiles(new Path("/"), true);
    while (locatedFileStatusRemoteIterator.hasNext()){
        LocatedFileStatus next = locatedFileStatusRemoteIterator.next();
        System.out.println(next.getPath().toString());
    }
    fileSystem.close();
}

下载文件到本地

程序执行的main方法

**
 * 拷贝文件的到本地
 * @throws Exception
 */
@Test
public void getFileToLocal()throws  Exception{
    FileSystem fileSystem = FileSystem.get(new URI("hdfs://192.168.47.100:8020"), new Configuration());
    FSDataInputStream open = fileSystem.open(new Path("/test/input/install.log"));
    FileOutputStream fileOutputStream = new FileOutputStream(new File("c:\\install.log"));
    IOUtils.copy(open,fileOutputStream );
    IOUtils.closeQuietly(open);
    IOUtils.closeQuietly(fileOutputStream);
    fileSystem.close();
}

hdfs上创建文件夹

@Test
public void mkdirs() throws  Exception{
    FileSystem fileSystem = FileSystem.get(new URI("hdfs://192.168.52.100:8020"), new Configuration());
    boolean mkdirs = fileSystem.mkdirs(new Path("/hello/mydir/test"));
    fileSystem.close();
}

hdfs文件上传

@Test
public void putData() throws  Exception{
    FileSystem fileSystem = FileSystem.get(new URI("hdfs://192.168.47.100:8020"), new Configuration());
    fileSystem.copyFromLocalFile(new Path("file:///c:\\install.log"),new Path("/hello/mydir/test"));
    fileSystem.close();
}

HDFS的小文件合并

由于hadoop擅长存储大文件,因为大文件的元数据信息比较少,如果hadoop集群当中有大量的小文件,那么每个小文件都需要维护一份元数据信息,会大大的增加集群管理元数据的内存压力,所以在实际工作当中,如果有必要一定要将小文件合并成大文件进行一起处理

在我们的hdfs 的shell命令模式下,可以通过命令行将很多的hdfs文件合并成一个大文件下载到本地,命令如下

cd /export/servers

hdfs dfs -getmerge /config/*.xml  ./hello.xml

既然可以在下载的时候将这些小文件合并成一个大文件一起下载,那么肯定就可以在上传的时候将小文件合并到一个大文件里面去

代码如下:

/**
 * 将多个本地系统文件,上传到hdfs,并合并成一个大的文件
 * @throws Exception
 */
@Test
public void mergeFile() throws  Exception{
    //获取分布式文件系统
    FileSystem fileSystem = FileSystem.get(new URI("hdfs://192.168.47.100:8020"), new Configuration(),"root");
    FSDataOutputStream outputStream = fileSystem.create(new Path("/bigfile.xml"));
    //获取本地文件系统
    LocalFileSystem local = FileSystem.getLocal(new Configuration());
    //通过本地文件系统获取文件列表,为一个集合
    FileStatus[] fileStatuses = local.listStatus(new Path("file:///F:\\上传小文件合并"));
    for (FileStatus fileStatus : fileStatuses) {
        FSDataInputStream inputStream = local.open(fileStatus.getPath());
       IOUtils.copy(inputStream,outputStream);
        IOUtils.closeQuietly(inputStream);
    }
    IOUtils.closeQuietly(outputStream);
    local.close();
    fileSystem.close();
}
(0)

相关推荐

  • HDFS的知识整理

    HDFS概述 HDFS产生背景 随着数据量越来越大,在一个操作系统存不下所有的数据,那么就分配到更多的操作系统管理的磁盘中,但是不方便管理和维护,迫切需要一种系统来管理多台机器上的文件,这就是分布式文 ...

  • 大数据开发之HDFS分布式文件存储系统详解

    HDFS(Hadoop Distributed File System)分布式文件存储系统,主要为各类分布式计算框架如Spark.MapReduce等提供海量数据存储服务,同时HBase.Hive底层 ...

  • 大数据开发之Yarn和Spark UI界面获取的方法

    一.Yarn 以获取Yarn界面队列信息为例: 1. 接口(HTTP Request) http://ip:port/ws/v1/cluster/scheduler ip和port:Yarn Reso ...

  • 大数据开发之Spark SQL/Hive实用函数分享

    字符串函数 1. concat 对字符串进行拼接:concat(str1, str2, ..., strN) ,参数:str1.str2...是要进行拼接的字符串. , (, ); 2. concat ...

  • 大数据开发之Spark入门

    什么是Spark? ·大数据的电花火石. ·Spark类似于MapReduce的低延迟的交互式计算框架. ·Spark是UC Berkeley AMPLab开发的是一种计算框架,分布式资源工作交由集群 ...

  • 大数据开发之Spark 基础入门学习

    集群相关 Cluster Manager指的是在集群上获取资源的外部服务,为每个spark application在集群中调度和分配资源的组件,目前有三种类型: Standalone:Spark 原生 ...

  • 大数据开发之Hadoop集群安装教程

    配置文件的修改 注意:以下所有操作都在node01主机进行. 1.1 hadoop-env.sh 1.介绍 文件中设置的是Hadoop运行时需要的环境变量.JAVA_HOME是必须设置的,即使我们当前 ...

  • 大数据开发之Hadoop生态6 大核心组件

    Hadoop 第一代分布式存储计算框架 Hadoop是一个分布式系统基础架构,由Apache基金会开发,它允许用户在不了解分布式底层细节的情况下,开发分布式程序,充分利用集群的威力高速运算和存储.Ha ...

  • 大数据开发之Hive应用分析

    Hive架构 1.用户接口:Client CLI(hive shell) JDBC(java访问hive) WEBUI(浏览器访问hive) 2.元数据(Metastore)-数据库 元数据包括:表名 ...

  • 大数据开发之Hive解析Json数组

    在Hive中会有很多数据是用Json格式来存储的,如开发人员对APP上的页面进行埋点时,会将多个字段存放在一个json数组中,因此数据平大数据培训台调用数据时,要对埋点数据进行解析.接下来就聊聊Hiv ...