HDFS应用开发篇

Posted

tags:

篇首语:本文由小常识网(cha138.com)小编为大家整理,主要介绍了HDFS应用开发篇相关的知识,希望对你有一定的参考价值。

HDFSjava操作

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

7.1 搭建开发环境

1、引入依赖

<dependency>

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

    <artifactId>hadoop-client</artifactId>

    <version>2.6.1</version>

</dependency>

 

注:如需手动引入jar包,hdfsjar----hadoop的安装目录的share

2window下开发的说明

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

A、windows的某个目录下解压一个hadoop的安装包

B、将安装包下的libbin目录用对应windows版本平台编译的本地库替换

C、window系统中配置HADOOP_HOME指向你解压的安装包

D、windows系统的path变量中加入hadoopbin目录

7.2 获取api中的客户端对象

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

Configuration conf = new Configuration()

FileSystem fs = FileSystem.get(conf)

 

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

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

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

 

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

 

 

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

技术分享 

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

hdfs dfsadmin -report查看状态

7.4.1 文件的增删改查

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("fs.defaultFS", "hdfs://hdp-node01:9000");

/**

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

 */

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

 

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

// fs = FileSystem.get(conf);

 

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

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

 

}

 

/**

 * hdfs上传文件

 *

 * @throws Exception

 */

@Test

public void testAddFileToHdfs() throws Exception {

 

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

Path src = new Path("g:/redis-recommend.zip");

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

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

fs.copyFromLocalFile(src, dst);

fs.close();

}

 

/**

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

 *

 * @throws IOException

 * @throws IllegalArgumentException

 */

@Test

public void testDownloadFileToLocal() throws IllegalArgumentException, IOException {

fs.copyToLocalFile(new Path("/jdk-7u65-linux-i586.tar.gz"), new Path("d:/"));

fs.close();

}

 

@Test

public void testMkdirAndDeleteAndRename() throws IllegalArgumentException, IOException {

 

// 创建目录

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

 

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

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

 

// 重命名文件或文件夹

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

 

}

 

/**

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

 *

 * @throws IOException

 * @throws IllegalArgumentException

 * @throws FileNotFoundException

 */

@Test

public void testListFiles() throws FileNotFoundException, IllegalArgumentException, IOException {

 

// 思考:为什么返回迭代器,而不是List之类的容器

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 bl : blockLocations) {

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

String[] hosts = bl.getHosts();

for (String host : hosts) {

System.out.println(host);

}

}

System.out.println("--------------angelababy打印的分割线--------------");

}

}

 

/**

 * 查看文件及文件夹信息

 *

 * @throws IOException

 * @throws IllegalArgumentException

 * @throws FileNotFoundException

 */

@Test

public void testListAll() throws FileNotFoundException, IllegalArgumentException, IOException {

 

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

}

}

}

 

 

 

7.4.2 通过流的方式访问hdfs

/**

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

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

 * @author

 *

 */

public class StreamAccess {


FileSystem fs = null;

 

@Before

public void init() throws Exception {

 

Configuration conf = new Configuration();

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

 

}




@Test

public void testDownLoadFileToLocal() throws IllegalArgumentException, IOException{


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

FSDataInputStream in = fs.open(new Path("/jdk-7u65-linux-i586.tar.gz"));


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

FileOutputStream out = new FileOutputStream(new File("c:/jdk.tar.gz"));


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

IOUtils.copyBytes(in, out, 4096);



}



/**

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

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

 * @throws IllegalArgumentException

 * @throws IOException

 */

@Test

public void testRandomAccess() throws IllegalArgumentException, IOException{

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

FSDataInputStream in = fs.open(new Path("/iloveyou.txt"));



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

in.seek(22);


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

FileOutputStream out = new FileOutputStream(new File("c:/iloveyou.line.2.txt"));


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


}




/**

 * 显示hdfs上文件的内容

 * @throws IOException

 * @throws IllegalArgumentException

 */

@Test

public void testCat() throws IllegalArgumentException, IOException{


FSDataInputStream in = fs.open(new Path("/iloveyou.txt"));


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

}

}

 

 

7.4.3 场景编程

mapreduce spark等运算框架中,有一个核心思想就是将运算移往数据,或者说,就是要在并发计算中尽可能让运算本地化,这就需要获取数据所在位置的信息并进行相应范围读取

以下模拟实现:获取一个文件的所有block位置信息,然后读取block中的内容

@Test

public void testCat() throws IllegalArgumentException, IOException{


FSDataInputStream in = fs.open(new Path("/weblog/input/access.log.10"));

//拿到文件信息

FileStatus[] listStatus = fs.listStatus(new Path("/weblog/input/access.log.10"));

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

BlockLocation[] fileBlockLocations = fs.getFileBlockLocations(listStatus[0], 0L, 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];


FileOutputStream os = new FileOutputStream(new File("d:/block0"));

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

os.write(b);

offset += 4096;

if(offset>=length) return;

};

os.flush();

os.close();

in.close();

}


本文出自 “上善若水” 博客,请务必保留此出处http://yushiwh.blog.51cto.com/2911795/1912405

以上是关于HDFS应用开发篇的主要内容,如果未能解决你的问题,请参考以下文章

Hadoop成长之路-HDFS基本操作篇

测试方法Flume+HDFS+Spark日志分析入门篇

大数据开发必备面试题Hadoop篇

HDFS源码学习纪录之-NameNode篇

Impala篇---Hue从初始到安装应用

Hive -- 原理篇