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秒。
需要注意的是hdfs-site.xml 配置文件中的heartbeat.recheck.interval的单位为毫秒，dfs.heartbeat.interval的单位为秒。所以，举个例子，如果heartbeat.recheck.interval设置为5000（毫秒），dfs.heartbeat.interval设置为3（秒，默认），则总的超时时间为40秒。
<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());}
}
}
通过流的方式访问hdfs：
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操作文件相关推荐
- Hadoop系列 (三)：HDFS详细介绍
文章目录 Hadoop系列文章 HDFS简介 HDFS基本概念 分布式文件系统特点 HDFS设计目标 HDFS架构 HDFS应用场景 HDFS三大组件 NameNode SecondaryNameNo ...
- Hadoop 核心编程之 HDFS 的文件操作
前言 本文并没有打算介绍 HDFS 的读写流程,虽然这是一块比较重要的内容.如果你感兴趣,可以去搜索相关资料.如果一遍没有看懂,请看第二遍. 本文还是以代码为主,并附以简短的说明,帮助你理解代码的逻辑 ...
- Hadoop之Hadoop企业优化（HDFS小文件优化）
Hadoop之Hadoop企业优化 目录 MapReduce 跑的慢的原因 MapReduce优化方法之数据输入 MapReduce优化方法之Map阶段 MapReduce优化方法之Reduce阶段 ...
- 三十四 操作文件和目录
如果我们要操作文件.目录,可以在命令行下面输入操作系统提供的各种命令来完成.比如dir.cp等命令. 如果要在Python程序中执行这些目录和文件的操作怎么办?其实操作系统提供的命令只是简单地调用了操 ...
- node用express写后端restful接口实战三：sequelize操作数据库：模型（Model）、迁移（Migration）与种子（Seeders）
一.创建模型与迁移 https://github.com/sequelize/cli 有了数据库,里面还要有 数据表.这次的项目非常简单,现在要建的表也只有一张,名字叫做 articles.使用代码操 ...
- python 新建文件 hdfs_python基础操作以及hdfs操作
一.前言 作为一个全栈工程师,必须要熟练掌握各种语言...HelloWorld.最近就被"逼着"走向了python开发之路,大体实现的功能是写一个通用类库将服务器本地存储的文件进行 ...
- 王家林 云计算分布式大数据Hadoop实战高手之路第七讲Hadoop图文训练课程：通过HDFS的心跳来测试replication具体的工作机制和流程...
这一讲主要深入使用HDFS命令行工具操作Hadoop分布式集群,主要是通过实验的配置hdfs-site.xml文件的心跳来测试replication具体的工作和流程. 通过HDFS的心跳来测试repl ...
- Azkaban实战，Command类型单一job示例，任务中执行外部shell脚本，Command类型多job工作flow,HDFS操作任务，MapReduce任务，HIVE任务
1.Azkaban实战 Azkaba内置的任务类型支持command.java Command类型单一job示例 1.创建job描述文件 vi command.job #command.job typ ...
- php对象存储hadoop存储,三个理由告诉你对象存储替换HDFS还不错
Hadoop使企业能够对庞大的非结构化数据集进行大规模分析处理.这个数据集可以包含数以百万计,甚至数十亿个需要读取的文件.为了降低成本并提高数据处理性能,数据和应用程序应该存在于相同的物理硬件上.这样 ...
最新文章
- OpenCV单kinect多帧静止场景的深度图像去噪
- Some notes for CLFS2017
- vc++ mfc中拖动效果的实现 借助于CImageList
- call,apply
- 关于std:auto_ptr
- 线程池之ThreadPool与ForkJoinPool
- 小书匠编辑器使用手册
- CentOS7安装MySQL及常用指令
- 网站项目建设流程概述
- SqlServer彻底卸载，适用于任何版本，亲测了n次都可用
- C语言图形库——easyx的使用
- bzoj5145 [Ynoi2018]未来日记 (多校第4场1013 Yuno and Claris)
- 中山大学生科院骆观正教授课题组诚聘博士后及副研究员
- mysql刷新当前数据库_sqlyog刷新数据库
- 【魏先生搞定Python系列】一文搞定Cufflinks画图
- 产品经理那些事：产品经理五年后的成长
- modis数据简介及批处理工具
- Linux Shell 使用手记
- python，抖音视频下载
- 2019全国大学生信息安全竞赛初赛writeup