本文主要是介绍hadoop实战(三) 使用HDFS操作文件，希望对大家解决编程问题提供一定的参考价值，需要的开发者们随着小编来一起学习吧！
一、hadoop简介
HADOOP集群具体来说包含两个集群：HDFS集群和YARN集群，两者逻辑上分离，但物理上常在一起
HDFS集群：负责海量数据的存储，集群中的角色主要有 NameNode / DataNode
YARN集群：负责海量数据运算时的资源调度，集群中的角色主要有 ResourceManager /NodeManager
二、hdfs的工作机制
(一)、概述
1. HDFS集群分为两大角色:NameNode、DataNode
2. NameNode负责管理整个文件系统的元数据
3. DataNode 负责管理用户的文件数据块
4. 文件会按照固定的大小（blocksize）切成若干块后分布式存储在若干台datanode上
5. 每一个文件块可以有多个副本，并存放在不同的datanode上
6. Datanode会定期向Namenode汇报自身所保存的文件block信息，而namenode则会负责保持文件的副本数量
7. HDFS的内部工作机制对客户端保持透明，客户端请求访问HDFS都是通过向namenode申请来进行
(二)、HDFS写数据流程
1、概述
客户端要向HDFS写数据，首先要跟namenode通信以确认可以写文件并获得接收文件block的datanode，然后，客户端按顺序将文件逐个block传递给相应datanode，并由接收到block的datanode负责向其他datanode复制block的副本2、详细步骤解析
1、根namenode通信请求上传文件，namenode检查目标文件是否已存在，父目录是否存在 2、namenode返回是否可以上传
3、client请求第一个 block该传输到哪些datanode服务器上
4、namenode返回3个datanode服务器ABC
5、client请求3台dn中的一台A上传数据（本质上是一个RPC调用，建立pipeline），A收到请求会继续调用B，然后B调用C，将整个pipeline建立完成，逐级返回客户端
6、client开始往A上传第一个block（先从磁盘读取数据放到一个本地内存缓存），以packet为单位，A收到一个packet就会传给B，B传给C；A每传一个packet会放入一个应答队列等待应答
7、当一个block传输完成之后，client再次请求namenode上传第二个block的服务器。
(三)、HDFS读数据流程
1、概述
客户端将要读取的文件路径发送给namenode，namenode获取文件的元信息（主要是block的存放位置信息）返回给客户端，客户端根据返回的信息找到相应datanode逐个获取文件的block并在客户端本地进行数据追加合并从而获得整个文件2、详细步骤解析
1、跟namenode通信查询元数据，找到文件块所在的datanode服务器 2、挑选一台datanode（就近原则，然后随机）服务器，请求建立socket流
3、datanode开始发送数据（从磁盘里面读取数据放入流，以packet为单位来做校验）
4、客户端以packet为单位接收，现在本地缓存，然后写入目标文件
(四)、NAMENODE工作机制
1、NAMENODE职责
负责客户端请求的响应
元数据的管理（查询，修改）
2、元数据管理
namenode对数据的管理采用了三种存储形式： （1）内存元数据(NameSystem)
（2）磁盘元数据镜像文件
（3）数据操作日志文件（可通过日志运算出元数据）
（1）元数据存储机制
A、内存中有一份完整的元数据(内存meta data) B、磁盘有一个“准完整”的元数据镜像（fsimage）文件(在namenode的工作目录中) C、用于衔接内存metadata和持久化元数据镜像fsimage之间的操作日志（edits文件）注：当客户端对hdfs中的文件进行新增或者修改操作，操作记录首先被记入edits日志文件中，当客户端操作成功后，相应的元数据会更新到内存meta.data中（2）元数据手动查看
可以通过hdfs的一个工具来查看edits中的信息
bin/hdfs oev -i edits -o edits.xml
bin/hdfs oiv -i fsimage_0000000000000000087 -p XML -o fsimage.xml
（3）元数据的checkpoint
每隔一段时间，会由secondary namenode将namenode上积累的所有edits和一个最新的fsimage下载到本地，并加载到内存进行merge（这个过程称为checkpoint） ##### checkpoint操作的触发条件配置参数:dfs.namenode.checkpoint.check.period=60 #检查触发条件是否满足的频率，60秒
dfs.namenode.checkpoint.dir=file://${hadoop.tmp.dir}/dfs/namesecondary
#以上两个参数做checkpoint操作时，secondary namenode的本地工作目录
dfs.namenode.checkpoint.edits.dir=${dfs.namenode.checkpoint.dir}dfs.namenode.checkpoint.max-retries=3 #最大重试次数
dfs.namenode.checkpoint.period=3600 #两次checkpoint之间的时间间隔3600秒
dfs.namenode.checkpoint.txns=1000000 #两次checkpoint之间最大的操作记录
checkpoint的附带作用:
namenode和secondary namenode的工作目录存储结构完全相同，所以，当namenode故障退出需要重新恢复时，可以从secondary namenode的工作目录中将fsimage拷贝到namenode的工作目录，以恢复namenode的元数据(五)、DATANODE的工作机制
1、概述
（1）Datanode工作职责：
存储管理用户的文件块数据
定期向namenode汇报自身所持有的block信息（通过心跳信息上报）（这点很重要，因为，当集群中发生某些block副本失效时，集群如何恢复block初始副本数量的问题）
<property><name>dfs.blockreport.intervalMsec</name><value>3600000</value><description>Determines block reporting interval in milliseconds.</description>
</property>
（2）Datanode掉线判断时限参数：
datanode进程死亡或者网络故障造成datanode无法与namenode通信，namenode不会立即把该节点判定为死亡，要经过一段时间，这段时间暂称作超时时长。HDFS默认的超时时长为10分钟+30秒。如果定义超时时间为timeout，则超时时长的计算公式为：timeout = 2 * heartbeat.recheck.interval + 10 * dfs.heartbeat.interval。
而默认的heartbeat.recheck.interval 大小为5分钟，dfs.heartbeat.interval默认为3秒。
<property><name>heartbeat.recheck.interval</name><value>2000</value>
</property>
<property><name>dfs.heartbeat.interval</name><value>1</value>
</property>
(六)、HDFS的java操作
hdfs在生产应用中主要是客户端的开发，其核心步骤是从hdfs提供的api中构造一个HDFS的访问客户端对象，然后通过该客户端对象操作（增删改查）HDFS上的文件
首先是依赖：
<dependency><groupId>org.apache.hadoop</groupId><artifactId>hadoop-client</artifactId><version>2.6.1</version></dependency><!--<dependency>--><!--<groupId>org.springframework.boot</groupId>--><!--<artifactId>spring-boot-starter-test</artifactId>--><!--<scope>test</scope>--><!--</dependency>--><dependency><groupId>junit</groupId><artifactId>junit</artifactId><version>4.10</version></dependency>
package com.hadoop.hdfs;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.*;
import org.junit.Before;
import org.junit.Test;import java.io.FileNotFoundException;
import java.io.IOException;
import java.net.URI;public class HdfsClient {FileSystem fs = null;@Before
public void init() throws Exception {// 构造一个配置参数对象，设置一个参数：我们要访问的hdfs的URI// 从而FileSystem.get()方法就知道应该是去构造一个访问hdfs文件系统的客户端，以及hdfs的访问地址// new Configuration();的时候，它就会去加载jar包中的hdfs-default.xml// 然后再加载classpath下的hdfs-site.xmlConfiguration conf = new Configuration();conf.set("fs.defaultFS", "hdfs://mini1:9000");/*** 参数优先级： 1、客户端代码中设置的值 2、classpath下的用户自定义配置文件 3、然后是服务器的默认配置*/conf.set("dfs.replication", "3");// 获取一个hdfs的访问客户端，根据参数，这个实例应该是DistributedFileSystem的实例// fs = FileSystem.get(conf);// 如果这样去获取，那conf里面就可以不要配"fs.defaultFS"参数，而且，这个客户端的身份标识已经是hadoop用户fs = FileSystem.get(new URI("hdfs://mini1:9000"), conf, "root");}
/*** 往hdfs上传文件** @throws Exception*/
@Test
public void testAddFileToHdfs() throws Exception {// 要上传的文件所在的本地路径Path src = new Path("/home/youjin/ruanjian/apache-tomcat-8.0.53.tar.gz");// 要上传到hdfs的目标路径Path dst = new Path("/apache-tomcat-8.0.53.tar.gz");fs.copyFromLocalFile(src, dst);fs.close();
}
/*** 从hdfs中复制文件到本地文件系统** @throws IOException* @throws IllegalArgumentException*/
@Test
public void testDownloadFileToLocal() throws IllegalArgumentException, IOException {fs.copyToLocalFile(new Path("/aaa"), new Path("/home/youjin/test"));fs.close();
}@Test
public void testMkdirAndDeleteAndRename() throws IllegalArgumentException, IOException {// 创建目录fs.mkdirs(new Path("/a1/b1/c1"));// 删除文件夹 ，如果是非空文件夹，参数2必须给值truefs.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());}
}
}
package com.hadoop.hdfs;import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.*;
import org.apache.hadoop.io.IOUtils;
import org.junit.Before;
import org.junit.Test;import java.io.File;
import java.io.FileOutputStream;
import java.io.IOException;
import java.net.URI;public class StreamAccess {FileSystem fs = null;@Before
public void init() throws Exception {Configuration conf = new Configuration();fs = FileSystem.get(new URI("hdfs://mini1:9000"), conf, "root");}
@Test
public void testDownLoadFileToLocal() throws IllegalArgumentException, IOException {//先获取一个文件的输入流----针对hdfs上的FSDataInputStream in = fs.open(new Path("/apache-tomcat-8.0.53.tar.gz"));//再构造一个文件的输出流----针对本地的FileOutputStream out = new FileOutputStream(new File("/home/youjin/test/apache-tomcat-8.0.53.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("/home/youjin/test/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);
}/*** 获取一个文件的所有block位置信息，然后读取指定block中的内容* @throws IllegalArgumentException* @throws IOException*/@Test
public void testCat1() throws IllegalArgumentException, IOException{FSDataInputStream in = fs.open(new Path("/wordcount/input/somewords.txt"));//拿到文件信息FileStatus[] listStatus = fs.listStatus(new Path("/wordcount/input/somewords.txt"));//获取这个文件的所有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("/home/youjin/test/block0"));while(in.read(offset, b, 0, 4096)!=-1){os.write(b);offset += 4096;if(offset>=length) return;};os.flush();os.close();in.close();
}
}
这篇关于hadoop实战(三) 使用HDFS操作文件的文章就介绍到这儿，希望我们推荐的文章对编程师们有所帮助！
