HDFS的API调用,创建Maven工程,创建一个非Maven工程,HDFS客户端操作数据代码示例,文件方式操作和流式操作

简介: 1. HDFS的java操作hdfs在生产应用中主要是客户端的开发,其核心步骤是从hdfs提供的api中构造一个HDFS的访问客户端对象,然后通过该客户端对象操作(增删改查)HDFS上的文件1.1 搭建开发环境                   1.1.1创建Maven工程快速创建一个Maven工程和目录结构的方式是执行下面的命令:mvn archetype:generate

1. HDFS的java操作

hdfs在生产应用中主要是客户端的开发,其核心步骤是从hdfs提供的api中构造一个HDFS的访问客户端对象,然后通过该客户端对象操作(增删改查)HDFS上的文件

1.1 搭建开发环境                  

1.1.1创建Maven工程

快速创建一个Maven工程和目录结构的方式是执行下面的命令:

mvn archetype:generate -DgroupId=com.toto.hadoop -DartifactId=hadoop -DarchetypeArtifactId=maven-archetype-quickstart -DinteractiveMode=false

接着执行下面的命令:

mvn -Pall eclipse:eclipse

mvn clean

mvn compile -Dmaven.test.skip=true

mvn install -Dmaven.test.skip=true

mvn package -Dmaven.test.skip=true

最后将工程导入到Eclipse中:

 

1、配置pom文件,引入下面的jar之后,其它相关的maven依赖的jar包,它会自动引入:

<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/maven-v4_0_0.xsd">

  <modelVersion>4.0.0</modelVersion>

  <groupId>com.toto.hadoop</groupId>

  <artifactId>hadoop</artifactId>

  <packaging>jar</packaging>

  <version>1.0-SNAPSHOT</version>

  <name>hadoop</name>

  <url>http://maven.apache.org</url>

  <dependencies>

        <dependency> 

            <groupId>org.apache.hadoop</groupId> 

            <artifactId>hadoop-common</artifactId> 

            <version>2.5.1</version> 

        </dependency> 

        <dependency> 

            <groupId>org.apache.hadoop</groupId> 

            <artifactId>hadoop-hdfs</artifactId> 

            <version>2.5.1</version> 

        </dependency> 

        <dependency> 

            <groupId>org.apache.hadoop</groupId> 

            <artifactId>hadoop-client</artifactId> 

            <version>2.5.1</version> 

        </dependency>

       

        <dependency>

           <groupId>junit</groupId>

           <artifactId>junit</artifactId>

           <version>4.12</version>

           <scope>test</scope>

       </dependency>

  </dependencies>

</project>

 

1.1.2创建一个非Maven工程

注:如需手动引入jar包,hdfs的jar包----hadoop的安装目录的share下,他们分别是:

hadoop-2.8.0\share\hadoop\common\hadoop-common-2.8.0.jar

hadoop-2.8.0\share\hadoop\common\lib

hadoop-2.8.0\share\hadoop\hdfs\hadoop-hdfs-2.8.0.jar

hadoop-2.8.0\share\hadoop\hdfs\lib

如果是创建非Maven工程,步骤如下:

第一步:创建一个User Library

第二步:添加library


第三步:添加common中的所有的jar


第四步:添加hadoop-hdfs-2.8.0.jar


第五步:添加hdfs依赖的jar


最后到工程里面查看引包情况


接着上面的配置,可以进行hdfs客户端调用的代码的编写了。

 

2、window下开发的说明

建议在linux下进行hadoop应用的开发,不会存在兼容性问题。如在window上做客户端应用开发,需要设置以下环境:

A、用老师给的windows平台下编译的hadoop安装包解压一份到windows的任意一个目录下

B、在window系统中配置HADOOP_HOME指向你解压的安装包目录

C、在windows系统的path变量中加入HADOOP_HOME的bin目录


1.2 获取api中的客户端对象

在java中操作hdfs,首先要获得一个客户端实例

Configuration conf = new Configuration()

FileSystem fs = FileSystem.get(conf)

 

而我们的操作目标是HDFS,所以获取到的fs对象应该是DistributedFileSystem的实例;

get方法是从何处判断具体实例化那种客户端类呢?

——从conf中的一个参数 fs.defaultFS的配置值判断;

 

如果我们的代码中没有指定fs.defaultFS,并且工程classpath下也没有给定相应的配置,conf中的默认值就来自于hadoop的jar包中的core-default.xml,默认值为: file:///,则获取的将不是一个DistributedFileSystem的实例,而是一个本地文件系统的客户端对象

 

1.3 DistributedFileSystem实例对象所具备的方法

1.4 HDFS客户端操作数据代码示例:

1.4.1 文件的增删改查

package com.toto.hadooptest;

 

import java.net.URI;

 

import org.apache.hadoop.conf.Configuration;

import org.apache.hadoop.fs.BlockLocation;

import org.apache.hadoop.fs.FileStatus;

import org.apache.hadoop.fs.FileSystem;

import org.apache.hadoop.fs.LocatedFileStatus;

import org.apache.hadoop.fs.Path;

import org.apache.hadoop.fs.RemoteIterator;

import org.junit.Before;

import org.junit.Test;

 

public class HdfsClient {

 

    FileSystem fs = null;

   

    @Before

    public void init() throws Exception {

        // 构造一个配置参数对象,设置一个参数:我们要访问的hdfsURI

        // 从而FileSystem.get()方法就知道应该是去构造一个访问hdfs文件系统的客户端,以及hdfs的访问地址

        // new Configuration();的时候,它就会去加载jar包中的hdfs-default.xml

        // 然后再加载classpath下的hdfs-site.xml

        Configuration conf = new Configuration();

        conf.set("df.defaultFS", "hdfs://hadoop:9000");

        /**

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

         */

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

       

        // 获取一个hdfs的访问客户端,根据参数,这个实例应该是DistributedFileSystem的实例

        // fs = FileSystem.get(conf);

       

        //如果这样去获取,那conf里面就可以不要配"fs.defaultFS"参数,而且,这个客户端的身份识别已经是hadoop用户

        //注意,最后的toto表示的是hadoop集群安装的Linux用户

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

    }

   

    /**

     * \brief hdfs上传文件

     *

     * 执行完成之后,进入Linux,执行一下命令,发现会出现以下列表

     * [toto@hadoop software]$ hadoop fs -ls /toto2

       Found 1 items

       -rw-r--r--   3 toto supergroup    2286088 2017-05-30 13:25 /toto2/ik-analyzer-solr5-master.zip

     */

    @Test

    public void testAddFileToHdfs() {

        try {

            //要上传的所在的本地路径

            Path src = new Path("d:/ik-analyzer-solr5-master.zip");

            //要上传的hdfs的目标路径

            Path dst = new Path("/toto2");

            fs.copyFromLocalFile(src, dst);

            fs.close();

        } catch (Exception e) {

            e.printStackTrace();

        }

    }

   

    /**

     * \brief hdfs中复制文件到本地文件系统

     * @attention执行完成之后,进入E:/learnTempFolder这个文件夹,发现在改文件夹下已经有了自己拷贝下来的文件了

     * @author toto

     * @date 2017530

     * @note  begin modify by 涂作权 2017530   原始创建

     */

    @Test

    public void testDownLoadFileToLocal() {

        try {

            fs.copyToLocalFile(new Path("/toto2/findbugs-1.3.9"), new Path("E:/learnTempFolder"));

            fs.close();

        } catch (Exception e) {

            e.printStackTrace();

        }

    }

   

    /**

     * 创建目录,删除目录,重命名

     * 执行之前hadoop的文件系统的目录结构如下:

     * [toto@hadoop sbin]$ hadoop fs -ls /

        Found 7 items

        drwxr-xr-x   - toto supergroup          0 2017-05-30 14:06 /aaa

        drwxr-xr-x   - toto supergroup          0 2017-05-29 14:01 /findbugs-1.3.9

        drwxr-xr-x   - toto supergroup          0 2017-05-29 03:23 /hive

        drwx------   - toto supergroup          0 2017-05-29 14:47 /tmp

        drwxr-xr-x   - toto supergroup          0 2017-05-29 23:59 /toto

        drwxr-xr-x   - toto supergroup          0 2017-05-30 13:25 /toto2

        drwxr-xr-x   - toto supergroup          0 2017-05-30 00:18 /user

        [toto@hadoop sbin]$

       

        执行之后的效果:

        [toto@hadoop sbin]$ hadoop fs -ls /

        Found 7 items

        drwxr-xr-x   - toto supergroup          0 2017-05-30 14:11 /a2

        drwxr-xr-x   - toto supergroup          0 2017-05-29 14:01 /findbugs-1.3.9

        drwxr-xr-x   - toto supergroup          0 2017-05-29 03:23 /hive

        drwx------   - toto supergroup          0 2017-05-29 14:47 /tmp

        drwxr-xr-x   - toto supergroup          0 2017-05-29 23:59 /toto

        drwxr-xr-x   - toto supergroup          0 2017-05-30 13:25 /toto2

        drwxr-xr-x   - toto supergroup          0 2017-05-30 00:18 /user

       

        总结:

        1、发现/aaa文件夹被删除了

        2、发现出现了/a2,但是最开始没有,说明是创建的a1,然后又被改成的a2

     */

    @Test

    public void testMkdirDeleteAndRename() {

        try {

            //创建目录

            fs.mkdirs(new Path("/a1/b1/c1"));

           

            //删除文件夹,如果是非空文件夹,参数2必须给值true

            fs.delete(new Path("/aaa"),true);

           

            //重命名文件或文件夹

            fs.rename(new Path("/a1"), new Path("/a2"));

        } catch(Exception e) {

            e.printStackTrace();

        }

    }

   

    /**

     * \brief 查看目录信息,只显示文件 

     */

    @Test

    public void testListFiles() {

        try {

            //思考:为什么返回迭代器,而不是List之类的容器,这里的只有调用hasNext()的时候,才会返回实际的数据信息

            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.getBlockSize());

                System.out.println(fileStatus.getPermission());

                System.out.println(fileStatus.getLen());

                BlockLocation[] blockLocations = fileStatus.getBlockLocations();

                for(BlockLocation b1 : blockLocations) {

                    System.out.println("block-length:" + b1.getLength() + "--" + "block-offset:" + b1.getOffset());

                    String[] hosts = b1.getHosts();

                    for(String host : hosts) {

                        System.out.println(host);

                    }

                }

                System.out.println("-----------------------------------");

            }

        } catch (Exception e) {

            e.printStackTrace();

        }

    }

   

    /**

     * 查看文件及文件夹信息

     *

     * 显示的内容如下:

     * d--             a2

     * d--             findbugs-1.3.9

     * d--             hive

     * d--             tmp

     * d--             toto

     * d--             toto2

     * d--             user

     */

    @Test

    public void testListAll() {

        try {

            FileStatus[] listStatus = fs.listStatus(new Path("/"));

           

            String flag = "d--             ";

            for(FileStatus fstatus : listStatus) {

                if (fstatus.isFile())  flag = "f--         ";

                System.out.println(flag + fstatus.getPath().getName());

            }

        } catch (Exception e) {

            e.printStackTrace();

        }

    }

}

 

1.4.2 通过流的方式访问hdfs

package com.toto.hadooptest;

 

import java.io.File;

import java.io.FileInputStream;

import java.io.FileOutputStream;

import java.net.URI;

 

import org.apache.hadoop.conf.Configuration;

import org.apache.hadoop.fs.BlockLocation;

import org.apache.hadoop.fs.FSDataInputStream;

import org.apache.hadoop.fs.FSDataOutputStream;

import org.apache.hadoop.fs.FileStatus;

import org.apache.hadoop.fs.FileSystem;

import org.apache.hadoop.fs.Path;

import org.apache.hadoop.io.IOUtils;

import org.junit.Before;

import org.junit.Test;

 

/**

 * 相对那些封装好的方法而言的更底层一些的操作方式

 * 上层那些mapreduce,spark等运算框架,去hdfs中获取数据的时候,就是调用这种底层的api

 */

public class StreamAccess {

    FileSystem fs = null;

 

    @Before

    public void init() throws Exception {

 

       Configuration conf = new Configuration();

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

 

    }

   

       /**

     * 通过流的方式上传文件到hdfs

     * @throws Exception

     */

    @Test

    public void testUpload() throws Exception {

       FSDataOutputStream outputStream = fs.create(new Path("/README.md"), true);

       FileInputStream inputStream = new FileInputStream("E:/learnTempFolder/dubbo-master/README.md");

      

       //注意这里的IOUtilsorg.apache.hadoop.io.IOUtils中的包,否则创建不成功

       IOUtils.copyBytes(inputStream, outputStream, 4096);

    }

 

    /**

     * hdfs上下载资源文件到本地

     */

    @Test

    public void testDownLoadFileToLocal() {

       try{

           //先获取一个文件的输入流---针对hdfs上的

           FSDataInputStream in = fs.open(new Path("/hive/apache-hive-2.0.0-bin.tar.gz"));

          

           //再构造一个文件的输出流---针对本地的

           FileOutputStream out = new FileOutputStream(new File("E:/apache-hive-2.0.0-bin.tar.gz"));

          

           //再将输入流中数据传输到输出流

           IOUtils.copyBytes(in, out, 4096);

       } catch (Exception e) {

           e.printStackTrace();

       }

    }

   

    /**

     * hdfs支持随机定位进行文件读取,而且可以方便地读取指定长度

     * 用于上层分布式运算框架并发处理数据

     */

    @Test

    public void testRandomAccess() {

       try {

           //先获取一个文件的输入流---针对hdfs上的

           FSDataInputStream in = fs.open(new Path("/pom.xml"));

          

           //可以将流的起始偏移量进行自定义

           in.seek(200);

          

           //在构造一个文件的输出流----针对本地的

           FileOutputStream out = new FileOutputStream(new File("E:/pom.xml"));

          

           //第三个参数的意思是获取长度为3000的文本内容

           IOUtils.copyBytes(in, out,3000L, true);

       } catch (Exception e) {

           e.printStackTrace();

       }

    }

   

    /**

     * 显示hdfs上文件的内容

     */

    @Test

    public void testCat() {

       try {

           FSDataInputStream in = fs.open(new Path("/pom.xml"));

          

           IOUtils.copyBytes(in, System.out, 1024);

       } catch (Exception e) {

           e.printStackTrace();

       }

    }

   

}

 

1.4.3 场景编程

在mapreduce 、spark等运算框架中,有一个核心思想就是将运算移往数据,或者说,就是要在并发计算中尽可能让运算本地化,这就需要获取数据所在位置的信息并进行相应范围读取以下模拟实现:获取一个文件的所有block位置信息,然后读取指定block中的内容

         @Test

    public void testCat2() {

       try {

           FSDataInputStream in = fs.open(new Path("/toto2/findbugs-1.3.9"));

           //拿到文件信息

           FileStatus[] listStatus = fs.listStatus(new Path("/toto2/findbugs-1.3.9"));

           //获取这个文件的所有的block的信息

           BlockLocation[] fileBlockLocations = fs.getFileBlockLocations(listStatus[0], 0, listStatus[0].getLen());

           //第一个block的长度

           long length = fileBlockLocations[0].getLength();

           //第一个block的起始偏移量

           long offset = fileBlockLocations[0].getOffset();

          

           System.out.println(length);

           System.out.println(offset);

          

           //获取第一个block写入输出流

           IOUtils.copyBytes(in, System.out, (int)length);

           byte[] b = new byte[4096];

          

           @SuppressWarnings("resource")

           FileOutputStream os = new FileOutputStream(new File("e:/block"));

           while(in.read(offset,b,0,4096) != -1) {

              os.write(b);

              offset += 4096;

              if (offset >= length) return;

           }

          

           os.flush();

           os.close();

           in.close();

       }  catch (Exception e) {

           e.printStackTrace();

       }

    }

 

 

 

 

目录
相关文章
|
1月前
|
Java API
掌握Java 8 Stream API的艺术:详解流式编程(三)
掌握Java 8 Stream API的艺术:详解流式编程
21 2
|
1月前
|
SQL JSON DataWorks
DataWorks操作报错合集之DataWorks报错显示API不存在,但这个API应该是有的,如何解决
DataWorks是阿里云提供的一站式大数据开发与治理平台,支持数据集成、数据开发、数据服务、数据质量管理、数据安全管理等全流程数据处理。在使用DataWorks过程中,可能会遇到各种操作报错。以下是一些常见的报错情况及其可能的原因和解决方法。
|
4天前
|
Java 大数据 API
【大数据】HDFS、HBase操作教程(含指令和JAVA API)
【大数据】HDFS、HBase操作教程(含指令和JAVA API)
33 0
【大数据】HDFS、HBase操作教程(含指令和JAVA API)
|
6天前
|
SQL 分布式计算 DataWorks
DataWorks操作报错合集之调用ListInstances API时报错:ServiceUnavailable The request has failed due to a temporary failure of the server.是什么导致的
DataWorks是阿里云提供的一站式大数据开发与治理平台,支持数据集成、数据开发、数据服务、数据质量管理、数据安全管理等全流程数据处理。在使用DataWorks过程中,可能会遇到各种操作报错。以下是一些常见的报错情况及其可能的原因和解决方法。
|
6天前
|
JavaScript 关系型数据库 Serverless
Serverless 应用引擎操作报错合集之在调用stable Diffusion函数计算api接口返回,报错:"Not Found"一般是什么原因
Serverless 应用引擎(SAE)是阿里云提供的Serverless PaaS平台,支持Spring Cloud、Dubbo、HSF等主流微服务框架,简化应用的部署、运维和弹性伸缩。在使用SAE过程中,可能会遇到各种操作报错。以下是一些常见的报错情况及其可能的原因和解决方法。
|
7天前
|
分布式计算 DataWorks API
DataWorks操作报错合集之如何解决API调用报400,文件夹找不到的错误
DataWorks是阿里云提供的一站式大数据开发与治理平台,支持数据集成、数据开发、数据服务、数据质量管理、数据安全管理等全流程数据处理。在使用DataWorks过程中,可能会遇到各种操作报错。以下是一些常见的报错情况及其可能的原因和解决方法。
|
10天前
|
安全 Java API
JavaSE——常用API进阶一(3/3)-StringBuilder(构造器、拼接内容、反转操作、其他操作),StringBuffer,StringJoiner
JavaSE——常用API进阶一(3/3)-StringBuilder(构造器、拼接内容、反转操作、其他操作),StringBuffer,StringJoiner
12 1
|
10天前
|
Java API
JavaSE——常用API进阶一(2/3)-Objects类(equals方法、isNull方法、nonNull方法)、包装类(介绍、包装方法及自动装拆箱、其他常见操作)
JavaSE——常用API进阶一(2/3)-Objects类(equals方法、isNull方法、nonNull方法)、包装类(介绍、包装方法及自动装拆箱、其他常见操作)
7 0
|
1月前
|
Linux API
Linux系统编程之文件编程常用API回顾和文件编程一般步骤
Linux系统编程之文件编程常用API回顾和文件编程一般步骤
Linux系统编程之文件编程常用API回顾和文件编程一般步骤
|
1月前
|
分布式计算 DataWorks 关系型数据库
DataWorks操作报错合集之在DataWorks同步数据时,遇到乱码问题,该怎么解决(rest api数据源)
DataWorks是阿里云提供的一站式大数据开发与治理平台,支持数据集成、数据开发、数据服务、数据质量管理、数据安全管理等全流程数据处理。在使用DataWorks过程中,可能会遇到各种操作报错。以下是一些常见的报错情况及其可能的原因和解决方法。
32 0

热门文章

最新文章

推荐镜像

更多