hdfs_hdfs shell - CSDN
hdfs 订阅
Hadoop分布式文件系统(HDFS)是指被设计成适合运行在通用硬件(commodity hardware)上的分布式文件系统(Distributed File System)。它和现有的分布式文件系统有很多共同点。但同时,它和其他的分布式文件系统的区别也是很明显的。HDFS是一个高度容错性的系统,适合部署在廉价的机器上。HDFS能提供高吞吐量的数据访问,非常适合大规模数据集上的应用。HDFS放宽了一部分POSIX约束,来实现流式读取文件系统数据的目的。HDFS在最开始是作为Apache Nutch搜索引擎项目的基础架构而开发的。HDFS是Apache Hadoop Core项目的一部分。HDFS有着高容错性(fault-tolerant)的特点,并且设计用来部署在低廉的(low-cost)硬件上。而且它提供高吞吐量(high throughput)来访问应用程序的数据,适合那些有着超大数据集(large data set)的应用程序。HDFS放宽了(relax)POSIX的要求(requirements)这样可以实现流的形式访问(streaming access)文件系统中的数据。 展开全文
Hadoop分布式文件系统(HDFS)是指被设计成适合运行在通用硬件(commodity hardware)上的分布式文件系统(Distributed File System)。它和现有的分布式文件系统有很多共同点。但同时,它和其他的分布式文件系统的区别也是很明显的。HDFS是一个高度容错性的系统,适合部署在廉价的机器上。HDFS能提供高吞吐量的数据访问,非常适合大规模数据集上的应用。HDFS放宽了一部分POSIX约束,来实现流式读取文件系统数据的目的。HDFS在最开始是作为Apache Nutch搜索引擎项目的基础架构而开发的。HDFS是Apache Hadoop Core项目的一部分。HDFS有着高容错性(fault-tolerant)的特点,并且设计用来部署在低廉的(low-cost)硬件上。而且它提供高吞吐量(high throughput)来访问应用程序的数据,适合那些有着超大数据集(large data set)的应用程序。HDFS放宽了(relax)POSIX的要求(requirements)这样可以实现流的形式访问(streaming access)文件系统中的数据。
信息
特    点
高容错性
适    用
大规模数据集
作    用
作为Apache Nutch的基础架构
实    质
分布式文件系统
全    称
Hadoop Distributed File System
简    称
hdfs
hdfs体系结构
HDFS采用了主从(Master/Slave)结构模型,一个HDFS集群是由一个NameNode和若干个DataNode组成的。其中NameNode作为主服务器,管理文件系统的命名空间和客户端对文件的访问操作;集群中的DataNode管理存储的数据。
收起全文
精华内容
参与话题
  • HDFS详解

    万次阅读 2017-07-10 12:19:57
    ******HDFS基本概念篇****** 1. HDFS前言 l 设计思想 分而治之:将大文件、大批量文件,分布式存放在大量服务器上,以便于采取分而治之的方式对海量数据进行运算分析;   l 在大数据系统中作用: 为各类分布式...
    ******HDFS基本概念篇******
    

    1. HDFS前言

    设计思想

    分而治之:将大文件、大批量文件,分布式存放在大量服务器上,以便于采取分而治之的方式对海量数据进行运算分析;

     

    在大数据系统中作用:

    为各类分布式运算框架(如:mapreducesparktez……)提供数据存储服务

     

    重点概念:文件切块,副本存放,元数据

     

    2. HDFS的概念和特性

    首先,它是一个文件系统,用于存储文件,通过统一的命名空间——目录树来定位文件

     

    其次,它是分布式的,由很多服务器联合起来实现其功能,集群中的服务器有各自的角色;

     

    重要特性如下:

    (1)HDFS中的文件在物理上是分块存储(block,块的大小可以通过配置参数( dfs.blocksize)来规定,默认大小在hadoop2.x版本中是128M,老版本中是64M

     

    (2)HDFS文件系统会给客户端提供一个统一的抽象目录树,客户端通过路径来访问文件,形如:hdfs://namenode:port/dir-a/dir-b/dir-c/file.data

     

    (3)目录结构及文件分块信息(元数据)的管理由namenode节点承担

    ——namenodeHDFS集群主节点,负责维护整个hdfs文件系统的目录树,以及每一个路径(文件)所对应的block块信息(blockid,及所在的datanode服务器)

     

     

    (4)文件的各个block的存储管理由datanode节点承担

    ---- datanodeHDFS集群从节点,每一个block都可以在多个datanode上存储多个副本(副本数量也可以通过参数设置dfs.replication

     

    (5)HDFS是设计成适应一次写入,多次读出的场景,且不支持文件的修改

     

    (注:适合用来做数据分析,并不适合用来做网盘应用,因为,不便修改,延迟大,网络开销大,成本太高)

    ******HDFS基本操作篇******

    3. HDFSshell(命令行客户端)操作

    3.1 HDFS命令行客户端使用

    HDFS提供shell命令行客户端,使用方法如下:

     

    3.2 命令行客户端支持的命令参数

            [-appendToFile <localsrc> ... <dst>]

            [-cat [-ignoreCrc] <src> ...]

            [-checksum <src> ...]

            [-chgrp [-R] GROUP PATH...]

            [-chmod [-R] <MODE[,MODE]... | OCTALMODE> PATH...]

            [-chown [-R] [OWNER][:[GROUP]] PATH...]

            [-copyFromLocal [-f] [-p] <localsrc> ... <dst>]

            [-copyToLocal [-p] [-ignoreCrc] [-crc] <src> ... <localdst>]

            [-count [-q] <path> ...]

            [-cp [-f] [-p] <src> ... <dst>]

            [-createSnapshot <snapshotDir> [<snapshotName>]]

            [-deleteSnapshot <snapshotDir> <snapshotName>]

            [-df [-h] [<path> ...]]

            [-du [-s] [-h] <path> ...]

            [-expunge]

            [-get [-p] [-ignoreCrc] [-crc] <src> ... <localdst>]

            [-getfacl [-R] <path>]

            [-getmerge [-nl] <src> <localdst>]

            [-help [cmd ...]]

            [-ls [-d] [-h] [-R] [<path> ...]]

            [-mkdir [-p] <path> ...]

            [-moveFromLocal <localsrc> ... <dst>]

            [-moveToLocal <src> <localdst>]

            [-mv <src> ... <dst>]

            [-put [-f] [-p] <localsrc> ... <dst>]

            [-renameSnapshot <snapshotDir> <oldName> <newName>]

            [-rm [-f] [-r|-R] [-skipTrash] <src> ...]

            [-rmdir [--ignore-fail-on-non-empty] <dir> ...]

            [-setfacl [-R] [{-b|-k} {-m|-x <acl_spec>} <path>]|[--set <acl_spec> <path>]]

            [-setrep [-R] [-w] <rep> <path> ...]

            [-stat [format] <path> ...]

            [-tail [-f] <file>]

            [-test -[defsz] <path>]

            [-text [-ignoreCrc] <src> ...]

            [-touchz <path> ...]

            [-usage [cmd ...]]

     

     

     

     

    3.2 常用命令参数介绍

    -help             

    功能:输出这个命令参数手册

    -ls                  

    功能:显示目录信息

    示例: hadoop fs -ls hdfs://hadoop-server01:9000/

    备注:这些参数中,所有的hdfs路径都可以简写

    -->hadoop fs -ls /   等同于上一条命令的效果

    -mkdir              

    功能:在hdfs上创建目录

    示例:hadoop fs  -mkdir  -p  /aaa/bbb/cc/dd

    -moveFromLocal            

    功能:从本地剪切粘贴到hdfs

    示例:hadoop  fs  - moveFromLocal  /home/hadoop/a.txt  /aaa/bbb/cc/dd

    -moveToLocal              

    功能:从hdfs剪切粘贴到本地

    示例:hadoop  fs  - moveToLocal   /aaa/bbb/cc/dd  /home/hadoop/a.txt

    --appendToFile  

    功能:追加一个文件到已经存在的文件末尾

    示例:hadoop  fs  -appendToFile  ./hello.txt  hdfs://hadoop-server01:9000/hello.txt

    可以简写为:

    Hadoop  fs  -appendToFile  ./hello.txt  /hello.txt

     

    -cat  

    功能:显示文件内容  

    示例:hadoop fs -cat  /hello.txt

     

    -tail                 

    功能:显示一个文件的末尾

    示例:hadoop  fs  -tail  /weblog/access_log.1

    -text                  

    功能:以字符形式打印一个文件的内容

    示例:hadoop  fs  -text  /weblog/access_log.1

    -chgrp

    -chmod

    -chown

    功能:linux文件系统中的用法一样,对文件所属权限

    示例:

    hadoop  fs  -chmod  666  /hello.txt

    hadoop  fs  -chown  someuser:somegrp   /hello.txt

    -copyFromLocal    

    功能:从本地文件系统中拷贝文件到hdfs路径去

    示例:hadoop  fs  -copyFromLocal  ./jdk.tar.gz  /aaa/

    -copyToLocal      

    功能:从hdfs拷贝到本地

    示例:hadoop fs -copyToLocal /aaa/jdk.tar.gz

    -cp              

    功能:从hdfs的一个路径拷贝hdfs的另一个路径

    示例: hadoop  fs  -cp  /aaa/jdk.tar.gz  /bbb/jdk.tar.gz.2

     

    -mv                     

    功能:在hdfs目录中移动文件

    示例: hadoop  fs  -mv  /aaa/jdk.tar.gz  /

    -get              

    功能:等同于copyToLocal,就是从hdfs下载文件到本地

    示例:hadoop fs -get  /aaa/jdk.tar.gz

    -getmerge             

    功能:合并下载多个文件

    示例:比如hdfs的目录 /aaa/下有多个文件:log.1, log.2,log.3,...

    hadoop fs -getmerge /aaa/log.* ./log.sum

    -put                

    功能:等同于copyFromLocal

    示例:hadoop  fs  -put  /aaa/jdk.tar.gz  /bbb/jdk.tar.gz.2

     

    -rm                

    功能:删除文件或文件夹

    示例:hadoop fs -rm -r /aaa/bbb/

     

    -rmdir                 

    功能:删除空目录

    示例:hadoop  fs  -rmdir   /aaa/bbb/ccc

    -df               

    功能:统计文件系统的可用空间信息

    示例:hadoop  fs  -df  -h  /

     

    -du

    功能:统计文件夹的大小信息

    示例:

    hadoop  fs  -du  -s  -h /aaa/*

     

    -count         

    功能:统计一个指定目录下的文件节点数量

    示例:hadoop fs -count /aaa/

     

    -setrep                

    功能:设置hdfs中文件的副本数量

    示例:hadoop fs -setrep 3 /aaa/jdk.tar.gz

    <这里设置的副本数只是记录在namenode的元数据中,是否真的会有这么多副本,还得看datanode的数量>

     

     

     

     

     

    ******HDFS原理篇******

    4. hdfs的工作机制

    (工作机制的学习主要是为加深对分布式系统的理解,以及增强遇到各种问题时的分析解决能力,形成一定的集群运维能力)

     

    注:很多不是真正理解hadoop技术体系的人会常常觉得HDFS可用于网盘类应用,但实际并非如此。要想将技术准确用在恰当的地方,必须对技术有深刻的理解

    4.1 概述

    1. HDFS集群分为两大角色:NameNodeDataNode

    2. NameNode负责管理整个文件系统的元数据

    3. DataNode 负责管理用户的文件数据块

    4. 文件会按照固定的大小(blocksize)切成若干块后分布式存储在若干台datanode

    5. 每一个文件块可以有多个副本,并存放在不同的datanode

    6. Datanode会定期向Namenode汇报自身所保存的文件block信息,而namenode则会负责保持文件的副本数量

    7. HDFS的内部工作机制对客户端保持透明,客户端请求访问HDFS都是通过向namenode申请来进行

     

     

    4.2 HDFS写数据流程

    4.2.1 概述

    客户端要向HDFS写数据,首先要跟namenode通信以确认可以写文件并获得接收文件blockdatanode,然后,客户端按顺序将文件逐个block传递给相应datanode,并由接收到blockdatanode负责向其他datanode复制block的副本

    4.2.2 详细步骤图

     

     

    4.2.3 详细步骤解析

    1、根namenode通信请求上传文件,namenode检查目标文件是否已存在,父目录是否存在

    2namenode返回是否可以上传

    3client请求第一个 block该传输到哪些datanode服务器上

    4namenode返回3datanode服务器ABC

    5client请求3dn中的一台A上传数据(本质上是一个RPC调用,建立pipeline),A收到请求会继续调用B,然后B调用C,将真个pipeline建立完成,逐级返回客户端

    6client开始往A上传第一个block(先从磁盘读取数据放到一个本地内存缓存),以packet为单位,A收到一个packet就会传给BB传给CA每传一个packet会放入一个应答队列等待应答

    7、当一个block传输完成之后,client再次请求namenode上传第二个block的服务器。

    4.3. HDFS读数据流程

    4.3.1 概述

    客户端将要读取的文件路径发送给namenodenamenode获取文件的元信息(主要是block的存放位置信息)返回给客户端,客户端根据返回的信息找到相应datanode逐个获取文件的block并在客户端本地进行数据追加合并从而获得整个文件

     

    4.3.2 详细步骤图

     

     

     

    4.3.3 详细步骤解析

    1、跟namenode通信查询元数据,找到文件块所在的datanode服务器

    2、挑选一台datanode(就近原则,然后随机)服务器,请求建立socket

    3datanode开始发送数据(从磁盘里面读取数据放入流,以packet为单位来做校验)

    4、客户端以packet为单位接收,现在本地缓存,然后写入目标文件

     

    5. NAMENODE工作机制

    学习目标:理解namenode的工作机制尤其是元数据管理机制,以增强对HDFS工作原理的理解,及培养hadoop集群运营中“性能调优”、“namenode”故障问题的分析解决能力

     

    问题场景:

    1、集群启动后,可以查看文件,但是上传文件时报错,打开web页面可看到namenode正处于safemode状态,怎么处理?

    2Namenode服务器的磁盘故障导致namenode宕机如何挽救集群及数据

    3Namenode是否可以有多个namenode内存要配置多大?namenode跟集群数据存储能力有关系吗?

    4、文件的blocksize究竟调大好还是调小好?

    ……

     

    诸如此类问题的回答,都需要基于对namenode自身的工作原理的深刻理解

     

    5.1 NAMENODE职责

    NAMENODE职责:

    负责客户端请求的响应

    元数据的管理(查询,修改)

    5.2 元数据管理

    namenode对数据的管理采用了三种存储形式:

    内存元数据(NameSystem)

    磁盘元数据镜像文件

    数据操作日志文件(可通过日志运算出元数据)

    5.2.1 元数据存储机制

    A、内存中有一份完整的元数据(内存meta data)

    B、磁盘有一个“准完整”的元数据镜像(fsimage文件(namenode的工作目录中)

    C、用于衔接内存metadata和持久化元数据镜像fsimage之间的操作日志(edits文件注:当客户端对hdfs中的文件进行新增或者修改操作,操作记录首先被记入edits日志文件中,当客户端操作成功后,相应的元数据会更新到内存meta.data

    5.2.2 元数据手动查看

    可以通过hdfs的一个工具来查看edits中的信息

    bin/hdfs oev -i edits -o edits.xml

    bin/hdfs oiv -i fsimage_0000000000000000087 -p XML -o fsimage.xml

    5.2.3 元数据的checkpoint

    每隔一段时间,会由secondary namenodenamenode上积累的所有edits和一个最新的fsimage下载到本地,并加载到内存进行merge(这个过程称为checkpoint

     

    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的附带作用

    namenodesecondary namenode的工作目录存储结构完全相同,所以,当namenode故障退出需要重新恢复时,可以从secondary namenode的工作目录中将fsimage拷贝到namenode的工作目录,以恢复namenode的元数据

     

     

    6. DATANODE的工作机制

    问题场景:

    1、集群容量不够,怎么扩容?

    2、如果有一些datanode宕机,该怎么办?

    3datanode明明已启动,但是集群中的可用datanode列表中就是没有,怎么办?

     

    以上这类问题的解答,有赖于对datanode工作机制的深刻理解

    6.1 概述

    1Datanode工作职责:

    存储管理用户的文件块数据

    定期向namenode汇报自身所持有的block信息(通过心跳信息上报)

    (这点很重要,因为,当集群中发生某些block副本失效时,集群如何恢复block初始副本数量的问题

     

    <property>

    <name>dfs.blockreport.intervalMsec</name>

    <value>3600000</value>

    <description>Determines block reporting interval in milliseconds.</description>

    </property>

     

    2Datanode掉线判断时限参数

    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>

     

     

    6.2 观察验证DATANODE功能

    上传一个文件,观察文件的block具体的物理存放情况:

     

    在每一台datanode机器上的这个目录中能找到文件的切块:

    /home/hadoop/app/hadoop-2.4.1/tmp/dfs/data/current/BP-193442119-192.168.2.120-1432457733977/current/finalized

     

     

     

     

    ******HDFS应用开发篇******

    7. 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客户端操作数据代码示例:

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

    }

     

     

    8. 案例1开发shell采集脚本

    8.1需求说明

    点击流日志每天都10T在业务应用服务器上,需要准实时上传数据仓库(Hadoop HDFS)上

    8.2需求分析

    一般上传文件都是在凌晨24点操作,由于很多种类的业务数据都要在晚上进行传输,为了减轻服务器的压力,避开高峰期

    如果需要伪实时的上传,则采用定时上传的方式

     

    8.3技术分析

     HDFS SHELL:  hadoop fs  –put   xxxx.tar  /data    还可以使用 Java Api

      满足上传一个文件,不能满足定时、周期性传入。

     定时调度器

    Linux crontab

    crontab -e

    */5 * * * * $home/bin/command.sh   //五分钟执行一次

    系统会自动执行脚本,每5分钟一次,执行时判断文件是否符合上传规则,符合则上传

    8.4实现流程

    8.4.1日志产生程序

    日志产生程序将日志生成后,产生一个一个的文件,使用滚动模式创建文件名。

     

     

    日志生成的逻辑由业务系统决定,比如在log4j配置文件中配置生成规则,如:xxxx.log 等于10G时,滚动生成新日志

    log4j.logger.msg=info,msg

    log4j.appender.msg=cn.maoxiangyi.MyRollingFileAppender

    log4j.appender.msg.layout=org.apache.log4j.PatternLayout

    log4j.appender.msg.layout.ConversionPattern=%m%n

    log4j.appender.msg.datePattern='.'yyyy-MM-dd

    log4j.appender.msg.Threshold=info

    log4j.appender.msg.append=true

    log4j.appender.msg.encoding=UTF-8

    log4j.appender.msg.MaxBackupIndex=100

    log4j.appender.msg.MaxFileSize=10GB

    log4j.appender.msg.File=/home/hadoop/logs/log/access.log

     

    细节

    1、 如果日志文件后缀是1\2\3等数字,该文件满足需求可以上传的话。把该文件移动到准备上传的工作区间。

    2、 工作区间有文件之后,可以使用hadoop put命令将文件上传。

    阶段问题:

    1、 待上传文件的工作区间的文件,在上传完成之后,是否需要删除掉。

    8.4.2伪代码

    使用ls命令读取指定路径下的所有文件信息,

    ls  | while read  line

     //判断line这个文件名称是否符合规则

    if  line=access.log.* (

    将文件移动到待上传的工作区间

    )

     

    //批量上传工作区间的文件

    hadoop fs  –put   xxx

     

     

    脚本写完之后,配置linux定时任务,每5分钟运行一次。

     

    8.5代码实现

    代码第一版本,实现基本的上传功能和定时调度功能

     

     

     

    代码第二版本:增强版V2(基本能用,还是不够健全)

     

     

    8.6效果展示及操作步骤

    1、日志收集文件收集数据,并将数据保存起来,效果如下:

     

    2、上传程序通过crontab定时调度

     

    3、程序运行时产生的临时文件

     

     

    4Hadoo hdfs上的效果

     

     

    9. 案例2:开发JAVA采集程序

    9.1 需求

    从外部购买数据,数据提供方会实时将数据推送到6FTP服务器上,我方部署6台接口采集机来对接采集数据,并上传到HDFS

     

    提供商在FTP上生成数据的规则是以小时为单位建立文件夹(2016-03-11-10),每分钟生成一个文件(00.dat,01.data,02.dat,........

     

    提供方不提供数据备份,推送到FTP服务器的数据如果丢失,不再重新提供,且FTP服务器磁盘空间有限,最多存储最近10小时内的数据

     

    由于每一个文件比较小,只有150M左右,因此,我方在上传到HDFS过程中,需要将15分钟时段的数据合并成一个文件上传到HDFS

     

    为了区分数据丢失的责任,我方在下载数据时最好进行校验

    9.2 设计分析

     

     

     

     

     


     

    展开全文
  • 深入理解Hadoop HDFS【一篇就够】

    万次阅读 多人点赞 2018-04-10 12:00:31
    https://blog.csdn.net/bingduanlbd/article/details/51914550#t24文本详细介绍了HDFS中的许多概念,对于理解Hadoop分布式文件系统很有帮助。1. 介绍在现代的企业环境中,单机容量往往无法存储大量数据,需要跨机器...

    又是一篇值得收藏温习的好文,更多好文请关注原文博主!!!

    原文=>https://blog.csdn.net/bingduanlbd/article/details/51914550#t24


    文本详细介绍了HDFS中的许多概念,对于理解Hadoop分布式文件系统很有帮助。

    1. 介绍

    在现代的企业环境中,单机容量往往无法存储大量数据,需要跨机器存储。统一管理分布在集群上的文件系统称为分布式文件系统。而一旦在系统中,引入网络,就不可避免地引入了所有网络编程的复杂性,例如挑战之一是如果保证在节点不可用的时候数据不丢失。

    传统的网络文件系统(NFS)虽然也称为分布式文件系统,但是其存在一些限制。由于NFS中,文件是存储在单机上,因此无法提供可靠性保证,当很多客户端同时访问NFS Server时,很容易造成服务器压力,造成性能瓶颈。另外如果要对NFS中的文件进行操作,需要首先同步到本地,这些修改在同步到服务端之前,其他客户端是不可见的。某种程度上,NFS不是一种典型的分布式系统,虽然它的文件的确放在远端(单一)的服务器上面。

    这里写图片描述

    这里写图片描述

    从NFS的协议栈可以看到,它事实上是一种VFS(操作系统对文件的一种抽象)实现。

    HDFS,是Hadoop Distributed File System的简称,是Hadoop抽象文件系统的一种实现。Hadoop抽象文件系统可以与本地系统、Amazon S3等集成,甚至可以通过Web协议(webhsfs)来操作。HDFS的文件分布在集群机器上,同时提供副本进行容错及可靠性保证。例如客户端写入读取文件的直接操作都是分布在集群各个机器上的,没有单点性能压力。

    如果你从零开始搭建一个完整的集群,参考[Hadoop集群搭建详细步骤(2.6.0)](http://blog.csdn.net/bingduanlbd/article/details/51892750

    2. HDFS设计原则

    HDFS设计之初就非常明确其应用场景,适用与什么类型的应用,不适用什么应用,有一个相对明确的指导原则。

    2.1 设计目标

    • 存储非常大的文件:这里非常大指的是几百M、G、或者TB级别。实际应用中已有很多集群存储的数据达到PB级别。根据Hadoop官网,Yahoo!的Hadoop集群约有10万颗CPU,运行在4万个机器节点上。更多世界上的Hadoop集群使用情况,参考Hadoop官网.

    • 采用流式的数据访问方式: HDFS基于这样的一个假设:最有效的数据处理模式是一次写入、多次读取数据集经常从数据源生成或者拷贝一次,然后在其上做很多分析工作 
      分析工作经常读取其中的大部分数据,即使不是全部。 因此读取整个数据集所需时间比读取第一条记录的延时更重要。

    • 运行于商业硬件上: Hadoop不需要特别贵的、reliable的(可靠的)机器,可运行于普通商用机器(可以从多家供应商采购) ,商用机器不代表低端机器。在集群中(尤其是大的集群),节点失败率是比较高的HDFS的目标是确保集群在节点失败的时候不会让用户感觉到明显的中断。

    2.2 HDFS不适合的应用类型

    有些场景不适合使用HDFS来存储数据。下面列举几个:

    1) 低延时的数据访问 
    对延时要求在毫秒级别的应用,不适合采用HDFS。HDFS是为高吞吐数据传输设计的,因此可能牺牲延时HBase更适合低延时的数据访问。

    2)大量小文件 
    文件的元数据(如目录结构,文件block的节点列表,block-node mapping)保存在NameNode的内存中, 整个文件系统的文件数量会受限于NameNode的内存大小。 
    经验而言,一个文件/目录/文件块一般占有150字节的元数据内存空间。如果有100万个文件,每个文件占用1个文件块,则需要大约300M的内存。因此十亿级别的文件数量在现有商用机器上难以支持。

    3)多方读写,需要任意的文件修改 
    HDFS采用追加(append-only)的方式写入数据。不支持文件任意offset的修改。不支持多个写入器(writer)。

    3. HDFS核心概念

    3.1 Blocks

    物理磁盘中有块的概念,磁盘的物理Block是磁盘操作最小的单元,读写操作均以Block为最小单元,一般为512 Byte。文件系统在物理Block之上抽象了另一层概念,文件系统Block物理磁盘Block的整数倍。通常为几KB。Hadoop提供的df、fsck这类运维工具都是在文件系统的Block级别上进行操作。

    HDFS的Block块比一般单机文件系统大得多,默认为128M。HDFS的文件被拆分成block-sized的chunk,chunk作为独立单元存储。比Block小的文件不会占用整个Block,只会占据实际大小。例如, 如果一个文件大小为1M,则在HDFS中只会占用1M的空间,而不是128M。

    HDFS的Block为什么这么大? 
    是为了最小化查找(seek)时间,控制定位文件与传输文件所用的时间比例。假设定位到Block所需的时间为10ms,磁盘传输速度为100M/s。如果要将定位到Block所用时间占传输时间的比例控制1%,则Block大小需要约100M。 
    但是如果Block设置过大,在MapReduce任务中,Map或者Reduce任务的个数 如果小于集群机器数量,会使得作业运行效率很低。

    Block抽象的好处 
    block的拆分使得单个文件大小可以大于整个磁盘的容量,构成文件的Block可以分布在整个集群, 理论上,单个文件可以占据集群中所有机器的磁盘。 
    Block的抽象也简化了存储系统,对于Block,无需关注其权限,所有者等内容(这些内容都在文件级别上进行控制)。 
    Block作为容错和高可用机制中的副本单元,即以Block为单位进行复制。

    3.2 Namenode & Datanode

    整个HDFS集群由Namenode和Datanode构成master-worker(主从)模式。Namenode负责构建命名空间,管理文件的元数据等,而Datanode负责实际存储数据,负责读写工作。

    Namenode

    Namenode存放文件系统树及所有文件、目录的元数据。元数据持久化为2种形式:

    • namespcae image
    • edit log

    但是持久化数据中不包括Block所在的节点列表,及文件的Block分布在集群中的哪些节点上,这些信息是在系统重启的时候重新构建(通过Datanode汇报的Block信息)。 
    在HDFS中,Namenode可能成为集群的单点故障,Namenode不可用时,整个文件系统是不可用的。HDFS针对单点故障提供了2种解决机制: 
    1)备份持久化元数据 
    将文件系统的元数据同时写到多个文件系统, 例如同时将元数据写到本地文件系统及NFS。这些备份操作都是同步的、原子的。

    2)Secondary Namenode 
    Secondary节点定期合并主Namenode的namespace image和edit log, 避免edit log过大,通过创建检查点checkpoint来合并。它会维护一个合并后的namespace image副本, 可用于在Namenode完全崩溃时恢复数据。下图为Secondary Namenode的管理界面:

    这里写图片描述

    Secondary Namenode通常运行在另一台机器,因为合并操作需要耗费大量的CPU和内存。其数据落后于Namenode,因此当Namenode完全崩溃时,会出现数据丢失。 通常做法是拷贝NFS中的备份元数据到Second,将其作为新的主Namenode。 
    在HA(High Availability高可用性)中可以运行一个Hot Standby,作为热备份,在Active Namenode故障之后,替代原有Namenode成为Active Namenode。

    Datanode

    数据节点负责存储和提取Block,读写请求可能来自namenode,也可能直接来自客户端。数据节点周期性向Namenode汇报自己节点上所存储的Block相关信息。

    3.3 Block Caching

    DataNode通常直接从磁盘读取数据,但是频繁使用的Block可以在内存中缓存。默认情况下,一个Block只有一个数据节点会缓存。但是可以针对每个文件可以个性化配置。 
    作业调度器可以利用缓存提升性能,例如MapReduce可以把任务运行在有Block缓存的节点上。 
    用户或者应用可以向NameNode发送缓存指令(缓存哪个文件,缓存多久), 缓存池的概念用于管理一组缓存的权限和资源。

    3.4 HDFS Federation

    我们知道NameNode的内存会制约文件数量,HDFS Federation提供了一种横向扩展NameNode的方式。在Federation模式中,每个NameNode管理命名空间的一部分,例如一个NameNode管理/user目录下的文件, 另一个NameNode管理/share目录下的文件。 
    每个NameNode管理一个namespace volumn,所有volumn构成文件系统的元数据。每个NameNode同时维护一个Block Pool,保存Block的节点映射等信息。各NameNode之间是独立的,一个节点的失败不会导致其他节点管理的文件不可用。 
    客户端使用mount table将文件路径映射到NameNode。mount table是在Namenode群组之上封装了一层,这一层也是一个Hadoop文件系统的实现,通过viewfs:协议访问。

    3.5 HDFS HA(High Availability高可用性)

    在HDFS集群中,NameNode依然是单点故障(SPOF: Single Point Of Failure)。元数据同时写到多个文件系统以及Second NameNode定期checkpoint有利于保护数据丢失,但是并不能提高可用性。 
    这是因为NameNode是唯一一个对文件元数据和file-block映射负责的地方, 当它挂了之后,包括MapReduce在内的作业都无法进行读写。

    当NameNode故障时,常规的做法是使用元数据备份重新启动一个NameNode。元数据备份可能来源于:

    • 多文件系统写入中的备份
    • Second NameNode的检查点文件

    启动新的Namenode之后,需要重新配置客户端和DataNode的NameNode信息。另外重启耗时一般比较久,稍具规模的集群重启经常需要几十分钟甚至数小时,造成重启耗时的原因大致有: 
    1) 元数据镜像文件载入到内存耗时较长。 
    2) 需要重放edit log 
    3) 需要收到来自DataNode的状态报告并且满足条件后才能离开安全模式提供写服务。

    Hadoop的HA方案

    采用HA的HDFS集群配置两个NameNode,分别处于Active和Standby状态。当Active NameNode故障之后,Standby接过责任继续提供服务,用户没有明显的中断感觉。一般耗时在几十秒到数分钟。 
    HA涉及到的主要实现逻辑有

    1) 主备需共享edit log存储。 
    主NameNode和待命的NameNode共享一份edit log,当主备切换时,Standby通过回放edit log同步数据。 
    共享存储通常有2种选择

    • NFS:传统的网络文件系统
    • QJM:quorum journal manager

    QJM是专门为HDFS的HA实现而设计的,用来提供高可用的edit log。QJM运行一组journal node,edit log必须写到大部分的journal nodes。通常使用3个节点,因此允许一个节点失败,类似ZooKeeper。注意QJM没有使用ZK,虽然HDFS HA的确使用了ZK来选举主Namenode。一般推荐使用QJM。

    2)DataNode需要同时往主备发送Block Report 
    因为Block映射数据存储在内存中(不是在磁盘上),为了在Active NameNode挂掉之后,新的NameNode能够快速启动,不需要等待来自Datanode的Block Report,DataNode需要同时向主备两个NameNode发送Block Report。

    3)客户端需要配置failover模式(失效备援模式,对用户透明) 

    Namenode的切换对客户端来说是无感知的,通过客户端库来实现。客户端在配置文件中使用的HDFS URI是逻辑路径,映射到一对Namenode地址。客户端会不断尝试每一个Namenode地址直到成功。

    4)Standby替代Secondary NameNode 
    如果没有启用HA,HDFS独立运行一个守护进程作为Secondary Namenode。定期checkpoint,合并镜像文件和edit日志。

    如果当主Namenode失败时,备份Namenode正在关机(停止 Standby),运维人员依然可以从头启动备份Namenode,这样比没有HA的时候更省事,算是一种改进,因为重启整个过程已经标准化到Hadoop内部,无需运维进行复杂的切换操作。

    NameNode的切换通过代failover controller来实现。failover controller有多种实现,默认实现使用ZooKeeper来保证只有一个Namenode处于active状态。

    每个Namenode运行一个轻量级的failover controller进程,该进程使用简单的心跳机制来监控Namenode的存活状态并在Namenode失败时触发failover。Failover可以由运维手动触发,例如在日常维护中需要切换主Namenode,这种情况graceful(优雅的) failover,非手动触发的failover称为ungraceful failover。

    在ungraceful failover的情况下,没有办法确定失败(被判定为失败)的节点是否停止运行,也就是说触发failover后,之前的主Namenode可能还在运行。QJM一次只允许一个Namenode写edit log,但是之前的主Namenode仍然可以接受读请求。Hadoop使用fencing来杀掉之前的Namenode。Fencing通过收回之前Namenode对共享的edit log的访问权限、关闭其网络端口使得原有的Namenode不能再继续接受服务请求。使用STONITH技术也可以将之前的主Namenode关机。

    最后,HA方案中Namenode的切换对客户端来说是不可见的,前面已经介绍过,主要通过客户端库来完成。

    4. 命令行接口

    HDFS提供了各种交互方式,例如通过Java API、HTTP、shell命令行的。命令行的交互主要通过hadoop fs来操作。例如:

    hadoop fs -copyFromLocal // 从本地复制文件到HDFS
    hadoop fs mkdir // 创建目录
    hadoop fs -ls  // 列出文件列表
    • 1
    • 2
    • 3

    Hadoop中,文件和目录的权限类似于POSIX模型,包括读、写、执行3种权限:

    • 读权限(r):用于读取文件或者列出目录中的内容
    • 写权限(w):对于文件,就是文件的写权限。目录的写权限指在该目录下创建或者删除文件(目录)的权限。
    • 执行权限(x):文件没有所谓的执行权限,被忽略。对于目录,执行权限用于访问器目录下的内容。

    每个文件或目录都有owner,group,mode三个属性,owner指文件的所有者,group为权限组。mode 
    由所有者权限、文件所属的组中组员的权限、非所有者非组员的权限组成。下图表示其所有者root拥有读写权限,supergroup组的组员有读权限,其他人有读权限。

    这里写图片描述

    文件权限是否开启通过dfs.permissions.enabled属性来控制,这个属性默认为false,没有打开安全限制,因此不会对客户端做授权校验,如果开启安全限制,会对操作文件的用户做权限校验。特殊用户superuser是Namenode进程的标识,不会针对该用户做权限校验。

    最后看一下ls命令的执行结果:

    这里写图片描述

    这个返回结果类似于Unix系统下的ls命令,第一栏为文件的mode,d表示目录,紧接着3种权限9位。 第二栏是指文件的副本数,这个数量通过dfs.replication配置,目录则使用-表示没有副本一说。其他诸如所有者、组、更新时间、文件大小跟Unix系统中的ls命令一致。

    如果需要查看集群状态或者浏览文件目录,可以访问Namenode暴露的Http Server查看集群信息,一般在namenode所在机器的50070端口。

    这里写图片描述

    这里写图片描述

    这里写图片描述

    5. Hadoop文件系统

    前面Hadoop的文件系统概念是抽象的,HDFS只是其中的一种实现。Hadoop提供的实现如下图:

    这里写图片描述 
    这里写图片描述

    简单介绍一下,Local是对本地文件系统的抽象,hdfs就是我们最常见的,两种web形式(webhdfs,swebhdfs)的实现通过HTTP提供文件操作接口。har是Hadoop体系下的压缩文件,当文件很多的时候可以压缩成一个大文件,可以有效减少元数据的数量。viewfs就是我们前面介绍HDFS Federation张提到的,用来在客户端屏蔽多个Namenode的底层细节。ftp顾名思义,就是使用ftp协议来实现,对文件的操作转化为ftp协议。s3a是对Amazon云服务提供的存储系统的实现,azure则是微软的云服务平台实现。

    前面我们提到了使用命令行跟HDFS交互,事实上还有很多方式来操作文件系统。例如Java应用程序可以使用org.apache.hadoop.fs.FileSystem来操作,其他形式的操作也都是基于FileSystem进行封装。我们这里主要介绍一下HTTP的交互方式。 
    WebHDFS和SWebHDFS协议将文件系统暴露HTTP操作,这种交互方式比原生的Java客户端慢,不适合操作大文件。通过HTTP,有2种访问方式,直接访问和通过代理访问

    直接访问 
    直接访问的示意图如下:

    这里写图片描述

    Namenode和Datanode默认打开了嵌入式web server,即dfs.webhdfs.enabled默认为true。webhdfs通过这些服务器来交互。元数据的操作通过namenode完成,文件的读写首先发到namenode,然后重定向到datanode读取(写入)实际的数据流。

    通过HDFS代理

    这里写图片描述

    采用代理的示意图如上所示。 使用代理的好处是可以通过代理实现负载均衡或者对带宽进行限制,或者防火墙设置。代理通过HTTP或者HTTPS暴露为WebHDFS,对应为webhdfs和swebhdfs URL Schema。

    代理作为独立的守护进程,独立于namenode和datanode,使用httpfs.sh脚本,默认运行在14000端口

    除了FileSystem直接操作,命令行,HTTTP外,还有C语言API,NFS,FUSER等方式,这里不做过多介绍。

    6. Java接口

    实际的应用中,对HDFS的大多数操作还是通过FileSystem来操作,这部分重点介绍一下相关的接口,主要关注HDFS的实现类DistributedFileSystem及相关类。

    6.1 读操作

    可以使用URL来读取数据,或者直接使用FileSystem操作。

    从Hadoop URL读取数据

    java.net.URL类提供了资源定位的统一抽象,任何人都可以自己定义一种URL Schema,并提供相应的处理类来进行实际的操作。hdfs schema便是这样的一种实现。

    InputStream in = null;
    try {
     in = new URL("hdfs://master/user/hadoop").openStream();
    }finally{
     IOUtils.closeStream(in);
    }
    • 1
    • 2
    • 3
    • 4
    • 5
    • 6

    为了使用自定义的Schema,需要设置URLStreamHandlerFactory,这个操作一个JVM只能进行一次,多次操作会导致不可用,通常在静态块中完成。下面的截图是一个使用示例:

    这里写图片描述

    这里写图片描述

    使用FileSystem API读取数据

    1) 首先获取FileSystem实例,一般使用静态get工厂方法

    public static FileSystem get(Configuration conf) throws IOException
    public static FileSystem get(URI uri , Configuration conf) throws IOException
    public static FileSystem get(URI uri , Configuration conf,String user) throws IOException
    • 1
    • 2
    • 3

    如果是本地文件,通过getLocal获取本地文件系统对象:

    public static LocalFileSystem getLocal(COnfiguration conf) thrown IOException
    • 1

    2)调用FileSystem的open方法获取一个输入流:

    public FSDataInputStream open(Path f) throws IOException
    public abstarct FSDataInputStream open(Path f , int bufferSize) throws IOException
    • 1
    • 2

    默认情况下,open使用4KB的Buffer,可以根据需要自行设置。

    3)使用FSDataInputStream进行数据操作 
    FSDataInputStream是java.io.DataInputStream的特殊实现,在其基础上增加了随机读取、部分读取的能力

    public class FSDataInputStream extends DataInputStream
        implements Seekable, PositionedReadable, 
          ByteBufferReadable, HasFileDescriptor, CanSetDropBehind, CanSetReadahead,
          HasEnhancedByteBufferAccess
    • 1
    • 2
    • 3
    • 4

    随机读取操作通过Seekable接口定义:

    public interface Seekable {
        void seek(long pos) throws IOException;
        long getPos() throws IOException;
    }
    • 1
    • 2
    • 3
    • 4

    seek操作开销昂贵,慎用。

    部分读取通过PositionedReadable接口定义:

    public interface PositionedReadable{
        public int read(long pistion ,byte[] buffer,int offser , int length) throws IOException;
        public int readFully(long pistion ,byte[] buffer,int offser , int length) throws IOException;
        public int readFully(long pistion ,byte[] buffer) throws IOException;
    }
    • 1
    • 2
    • 3
    • 4
    • 5

    6.2 写数据

    在HDFS中,文件使用FileSystem类的create方法及其重载形式来创建,create方法返回一个输出流FSDataOutputStream,可以调用返回输出流的getPos方法查看当前文件的位移,但是不能进行seek操作,HDFS仅支持追加操作。

    创建时,可以传递一个回调接口Peofressable,获取进度信息

    append(Path f)方法用于追加内容到已有文件,但是并不是所有的实现都提供该方法,例如Amazon的文件实现就没有提供追加功能。

    下面是一个例子:

    String localSrc =  args[0];
    String dst = args[1];
    
    InputStream in = new BufferedInputStream(new FileInputStream(localSrc));
    
    Configuration conf = new Configuration();
    FileSystem fs = FileSystem.get(URI.create(dst),conf);
    
    OutputStream out = fs.create(new Path(dst), new Progressable(){
        public vid progress(){
            System.out.print(.);
        }
    });
    
    IOUtils.copyBytes(in , out, 4096,true);
    • 1
    • 2
    • 3
    • 4
    • 5
    • 6
    • 7
    • 8
    • 9
    • 10
    • 11
    • 12
    • 13
    • 14
    • 15

    6.3 目录操作

    使用mkdirs()方法,会自动创建没有的上级目录

    HDFS中元数据封装在FileStatus类中,包括长度、block size,replicaions,修改时间、所有者、权限等信息。使用FileSystem提供的getFileStatus方法获取FileStatus。exists()方法判断文件或者目录是否存在;

    列出文件(list),则使用listStatus方法,可以查看文件或者目录的信息

      public abstract FileStatus[] listStatus(Path f) throws FileNotFoundException, 
                                                             IOException;
    • 1
    • 2

    Path是个文件的时候,返回长度为1的数组。FileUtil提供的stat2Paths方法用于将FileStatus转化为Path对象。

    globStatus则使用通配符对文件路径进行匹配:

    public FileStatus[] globStatus(Path pathPattern) throws IOException
    • 1

    PathFilter用于自定义文件名过滤,不能根据文件属性进行过滤,类似于java.io.FileFilter。例如下面这个例子排除到给定正则表达式的文件:

    public interfacePathFilter{
        boolean accept(Path path);
    }
    • 1
    • 2
    • 3

    6.4 删除数据

    使用FileSystem的delete()方法

    public boolean delete(Path f , boolean recursive) throws IOException;
    • 1

    recursive参数在f是个文件的时候被忽略。如果f是文件并且recursice为true,则删除整个目录,否则抛出异常.

    7. 数据流(读写流程)

    接下来详细介绍HDFS读写数据的流程,以及一致性模型相关的一些概念。

    7.1 读文件

    大致读文件的流程如下:

    这里写图片描述

    1)客户端传递一个文件Path给FileSystem的open方法

    2)DFS采用RPC远程获取文件最开始的几个block的datanode地址。Namenode会根据网络拓扑结构决定返回哪些节点(前提是节点有block副本),如果客户端本身是Datanode并且节点上刚好有block副本,直接从本地读取。

    3)客户端使用open方法返回的FSDataInputStream对象读取数据(调用read方法)

    4)DFSInputStream(FSDataInputStream实现了改类)连接持有第一个block的、最近的节点,反复调用read方法读取数据

    5)第一个block读取完毕之后,寻找下一个block的最佳datanode,读取数据。如果有必要,DFSInputStream会联系Namenode获取下一批Block 的节点信息(存放于内存,不持久化),这些寻址过程对客户端都是不可见的。

    6)数据读取完毕,客户端调用close方法关闭流对象

    在读数据过程中,如果与Datanode的通信发生错误,DFSInputStream对象会尝试从下一个最佳节点读取数据,并且记住该失败节点, 后续Block的读取不会再连接该节点 
    读取一个Block之后,DFSInputStram会进行检验和验证,如果Block损坏,尝试从其他节点读取数据,并且将损坏的block汇报给Namenode。 
    客户端连接哪个datanode获取数据,是由namenode来指导的,这样可以支持大量并发的客户端请求,namenode尽可能将流量均匀分布到整个集群。 
    Block的位置信息是存储在namenode的内存中,因此相应位置请求非常高效,不会成为瓶颈。

    7.2 写文件

    这里写图片描述

    步骤分解 
    1)客户端调用DistributedFileSystem的create方法

    2)DistributedFileSystem远程RPC调用Namenode在文件系统的命名空间中创建一个新文件,此时该文件没有关联到任何block。 这个过程中,Namenode会做很多校验工作,例如是否已经存在同名文件,是否有权限,如果验证通过,返回一个FSDataOutputStream对象。 如果验证不通过,抛出异常到客户端。

    3)客户端写入数据的时候,DFSOutputStream分解为packets(数据包),并写入到一个数据队列中,该队列由DataStreamer消费。

    4)DateStreamer负责请求Namenode分配新的block存放的数据节点。这些节点存放同一个Block的副本,构成一个管道。 DataStreamer将packet写入到管道的第一个节点,第一个节点存放好packet之后,转发给下一个节点,下一个节点存放 之后继续往下传递。

    5)DFSOutputStream同时维护一个ack queue队列,等待来自datanode确认消息。当管道上的所有datanode都确认之后,packet从ack队列中移除。

    6)数据写入完毕,客户端close输出流。将所有的packet刷新到管道中,然后安心等待来自datanode的确认消息。全部得到确认之后告知Namenode文件是完整的。 Namenode此时已经知道文件的所有Block信息(因为DataStreamer是请求Namenode分配block的),只需等待达到最小副本数要求,然后返回成功信息给客户端。

    Namenode如何决定副本存在哪个Datanode?

    HDFS的副本的存放策略是可靠性、写带宽、读带宽之间的权衡。默认策略如下:

    • 第一个副本放在客户端相同的机器上,如果机器在集群之外,随机选择一个(但是会尽可能选择容量不是太慢或者当前操作太繁忙的)
    • 第二个副本随机放在不同于第一个副本的机架上。
    • 第三个副本放在跟第二个副本同一机架上,但是不同的节点上,满足条件的节点中随机选择。
    • 更多的副本在整个集群上随机选择,虽然会尽量避免太多副本在同一机架上。 
      副本的位置确定之后,在建立写入管道的时候,会考虑网络拓扑结构。下面是可能的一个存放策略:

    这里写图片描述

    这样选择很好滴平衡了可靠性、读写性能

    • 可靠性:Block分布在两个机架上
    • 写带宽:写入管道的过程只需要跨越一个交换机
    • 读带宽:可以从两个机架中任选一个读取

    7.3 一致性模型

    一致性模型描述文件系统中读写操纵的可见性。HDFS中,文件一旦创建之后,在文件系统的命名空间中可见:

    Path p = new Path("p");
    fs.create(p);
    assertTaht(fs.exists(p),is(true));
    • 1
    • 2
    • 3

    但是任何被写入到文件的内容不保证可见,即使对象流已经被刷新。 
    “`java 
    Path p = new Path(“p”); 
    OutputStream out = fs.create(p); 
    out.write(“content”.getBytes(“UTF-8”)); 
    out.flush(); 
    assertTaht(fs.getFileStatus(p).getLen,0L); // 为0,即使调用了flush

    
    如果需要强制刷新数据到Datanode,使用FSDataOutputStream的hflush方法强制将缓冲刷到datanode
    hflush之后,HDFS保证到这个时间点为止写入到文件的数据都到达所有的数据节点。
     ```java
    Path p = new Path("p");
    OutputStream out = fs.create(p);
    out.write("content".getBytes("UTF-8"));
    out.flush();
    assertTaht(fs.getFileStatus(p).getLen,is(((long,"content".length())));
    • 1
    • 2
    • 3
    • 4
    • 5
    • 6
    • 7
    • 8
    • 9

    关闭对象流时,内部会调用hflush方法,但是hflush不保证datanode数据已经写入到磁盘,只是保证写入到datanode的内存, 因此在机器断电的时候可能导致数据丢失,如果要保证写入磁盘,使用hsync方法,hsync类型与fsync()的系统调用,fsync提交某个文件句柄的缓冲数据。

    FileOutputStreamout = new FileOutPutStream(localFile);
    out.write("content".getBytes("UTF-8"));
    out.flush();
    out.getFD().sync();
    assertTaht(localFile.getLen,is(((long,"content".length())));
    • 1
    • 2
    • 3
    • 4
    • 5

    使用hflush或hsync会导致吞吐量下降,因此设计应用时,需要在吞吐量以及数据的健壮性之间做权衡。

    另外,文件写入过程中,当前正在写入的Block对其他Reader不可见。

    7.4 Hadoop节点距离

    在读取和写入的过程中,namenode在分配Datanode的时候,会考虑节点之间的距离。HDFS中,距离没有 
    采用带宽来衡量,因为实际中很难准确度量两台机器之间的带宽。 
    Hadoop把机器之间的拓扑结构组织成树结构,并且用到达公共父节点所需跳转数之和作为距离。事实上这是一个距离矩阵的例子。下面的例子简明地说明了距离的计算:

    这里写图片描述

    同一数据中心,同一机架,同一节点距离为0

    同一数据中心,同一机架,不同节点距离为2

    同一数据中心,不同机架,不同节点距离为4

    不同数据中心,不同机架,不同节点距离为6

    这里写图片描述

    Hadoop集群的拓扑结构需要手动配置,如果没配置,Hadoop默认所有节点位于同一个数据中心的同一机架上。

    8 相关运维工具

    8.1 使用distcp并行复制

    前面的关注点都在于单线程的访问,如果需要并行处理文件,需要自己编写应用。Hadoop提供的distcp工具用于并行导入数据到Hadoop或者从Hadoop导出。一些例子:

    hadoop distcp file1 file2  //可以作为fs -cp命令的高效替代
    hadoop distcp dir1 dir2
    hadoop distcp -update dir1 dir2 #update参数表示只同步被更新的文件,其他保持不变
    • 1
    • 2
    • 3

    distcp是底层使用MapReduce实现,只有map实现,没有reduce。在map中并行复制文件。 distcp尽可能在map之间平均分配文件。map的数量可以通过-m参数指定:

    hadoop distcp -update -delete -p hdfs://master1:9000/foo hdfs://master2/foo 
    • 1

    这样的操作常用于在两个集群之间复制数据,update参数表示只同步被更新过的数据,delete会删除目标目录中存在,但是源目录不存在的文件。p参数表示保留文件的全校、block大小、副本数量等属性。

    如果两个集群的Hadoop版本不兼容,可以使用webhdfs协议:

    hadoop distcp webhdfs://namenode1:50070/foo webhdfs://namenode2:50070/foo
    • 1

    8.2 平衡HDFS集群

    在distcp工具中,如果我们指定map数量为1,不仅速度很慢,每个Block第一个副本将全部落到运行这个唯一map的节点上,直到磁盘溢出。因此使用distcp的时候,最好使用默认的map数量,即20. 
    HDFS在Block均匀分布在各个节点上的时候工作得最好,如果没有办法在作业中尽量保持集群平衡,例如为了限制map数量(以便其他节点可以被别的作业使用),那么可以使用balancer工具来调整集群的Block分布。

    参考

    主要参考《Hadoop》权威指南第3章,自己进一步整理。感谢原作者提供这么好的书籍。

    展开全文
  • HDFS入门简介

    千次阅读 2019-06-21 14:12:08
    HDFS是什么? 易于扩展的分布式文件系统 运行在大量普通廉价机器上提供容错机制 为大量用户提供性能不错的存取服务 设计目标: 自动快速检测应对硬件错误 流式访问数据,以流的方式访问数据,设计用于数据的批量...

    HDFS是什么?

    1. 易于扩展的分布式文件系统
    2. 运行在大量普通廉价机器上提供容错机制
    3. 为大量用户提供性能不错的存取服务

    设计目标:

    1. 自动快速检测应对硬件错误
    2. 流式访问数据,以流的方式访问数据,设计用于数据的批量处理
      缺点:不适合存储大量小文件;不适合低延迟的数据访问;不支持多用户写入及任意修改文件
    3. 移动计算不移动数据(大数据基本原则,空间换时间)
    4. 简单一致性模型
    5. 异构平台可移植性

    安装配置

    追加连接

    hdfs://(协议头)host(主机名):port(端口号)/
    

    查看${HADOOP_HOME}/bin${HADOOP_HOME}/sbin
    在这里插入图片描述

    #学会help(很多地方都可以查看帮助文档)
    hdfs dfsadmin -help
    

    HDFS的基本组件

    1. namenode:管理整个文件系统的元数据
      namenode工作职责:管理元数据、维护目录结构、响应客户端请求
    2. datanode:复制管理用户的文件数据块
      datanode的工作职责:管理用户提交的数据 心跳机制 块报告
    3. secondarynamenode:namenode的助理,帮助加载元数据,紧急情况下(例如namenode宕机),可以帮助恢复数据
      在这里插入图片描述

    HDFS读写流程图解

    一、 写数据流程
    在这里插入图片描述
    节点服务器传输数据方式:网络传输,以package包的形式(第8步,上传数据的时候会把package先放到缓存队列,如果此时package出错的话,会默认重传 4次)
    这里的话,追加一下一些问题(分布式系统之间可能故障,而且网络的不可靠性都是设计人员需要考虑的问题):socket(长连接),http(短连接),还有其他的方式,比如管道、FIFO、消息队列(kafka。。。)
    为什么使用长链接?
    最简单分布式系统是一直存在的,很少是短时间的访问,维持心跳机制
    什么是心跳机制?
    namenode启动的时候,会有一个加载元数据(数据的数据,类似于表的索引)和块报告(datanode会定时(可以再配置文件中设置,所以一定要时间同步)对块信息进行统计)的过程,namenode通过心跳机制维护整个集群的可用性。如果块报告上传失败,namenode不会更新元数据,在块报告的时候就会将其删除掉。
    安全模式
    什么时候进入安全模式?
    刚刚启动(namenode加载元数据的时候(先加载元数据镜像到内存中,在将edits日志的操作在内存中执行一遍,namenode进入安全模式,进行块报告,阈值安全的话30秒退出安全模式))
    阈值低于0.999f(默认)
    datanode存活数量小于0

    怎么解除安全模式?
    1. 格式化集群(需要删除namenode.dir的配置路径)
    基本不会采用这种方式
    2. 强制离开安全模式
    hdfs dfsadmin -safemode leave
    3 .
    # 检测集群文件、节点、块是否出现问题
    hdfs fsck /
    #删除损坏块的block
    hdfs fsck / -delete
    4. 调低阈值(在配置文件 safemode)
    二、 读数据流程
    在这里插入图片描述
    磁盘故障
    多个副本策略
    namenode故障宕机
    简单方案:secondarynamenode取出fsimage文件copy到namenode的元数据存储目录下
    完美解决:在namenode上挂多块磁盘,配置fs.namenode.name.dir(用,分割磁盘 )

    CheckPoint

    触发条件:

    1. 事务达到1000000条(默认)
    2. 1小时(默认)
    <property>
      <name>dfs.namenode.checkpoint.dir</name>
      <value>/hadoop/data/name</value>
    </property>
    <!--日志文件edits的检测目录-->
    <property>
      <name>dfs.namenode.checkpoint.edits.dir</name>
      <value>/hadoop/data/edits</value>
    </property>
    <!--时间一小时-->
    <property>
      <name>dfs.namenode.checkpoint.period</name>
      <value>3600</value>
    </property>
    <!--事物达到1000000-->
    <property>
      <name>dfs.namenode.checkpoint.txns</name>
      <value>1000000</value>
    </property>
    

    在这里插入图片描述
    注:namenode存储元数据,secondarynamenode执行checkpoint的时候去namenode下载edits和fsimage
    注意的问题
    客户端和服务器端解释

    1. 客户端的配置文件决定副本数量,而不是服务器
    2. 文件存储以块的形式存储在服务器上(客户端决定文件切分,块大小)
    展开全文
  • HDFS全面详解

    万次阅读 多人点赞 2019-08-14 10:04:07
    文章目录课程大纲(HDFS详解)学习目标:HDFS基本概念篇1.1HDFS前言1.2HDFS的概念和特性HDFS基本操作篇2.1HDFS的shell(命令行客户端)操作2.1.1 HDFS命令行客户端使用2.2 命令行客户端支持的命令参数2.3 常用命令参数...

    课程大纲(HDFS详解)

    Hadoop HDFS 分布式文件系统DFS简介
    HDFS的系统组成介绍
    HDFS的组成部分详解
    副本存放策略及路由规则
    命令行接口
    Java接口
    客户端与HDFS的数据流讲解

    学习目标:

    掌握hdfs的shell操作
    掌握hdfs的java api操作
    理解hdfs的工作原理


    HDFS基本概念篇

    1.1HDFS前言

    • 设计思想
      分而治之:将大文件、大批量文件,分布式存放在大量服务器上,以便于采取分而治之的方式对海量数据进行运算分析;

    • 在大数据系统中作用:
      为各类分布式运算框架(如:mapreduce,spark,tez,……)提供数据存储服务

    • 重点概念:文件切块,副本存放,元数据
      补充:
      hdfs是架在本地文件系统上面的分布式文件系统,它就是个软件,也就是用一套代码把底下所有机器的硬盘变成一个软件下的目录,和mysql没有什么区别,思想一样。
      mysql 本质是一个解析器,把sql变成io去读文件,再把数据转换出来给用户,存文件的底层就是使用linux或者windows的文件系统,文件名就是表名,目录名就是库名。

    1.2HDFS的概念和特性

    首先,它是一个文件系统,用于存储文件,通过统一的命名空间——目录树来定位文件

    其次,它是分布式的,由很多服务器联合起来实现其功能,集群中的服务器有各自的角色;

    重要特性如下:
    (1)HDFS中的文件在物理上是分块存储(block),块的大小可以通过配置参数( dfs.blocksize)来规定,默认大小在hadoop2.x版本中是128M,老版本中是64M

    (2)HDFS文件系统会给客户端提供一个统一的抽象目录树,客户端通过路径来访问文件,形如:hdfs://namenode:port/dir-a/dir-b/dir-c/file.data

    (3)**目录结构及文件分块信息(元数据)**的管理由namenode节点承担
    ——namenode是HDFS集群主节点,负责维护整个hdfs文件系统的目录树,以及每一个路径(文件)所对应的block块信息(block的id,及所在的datanode服务器)

    (4)文件的各个block的存储管理由datanode节点承担
    ---- datanode是HDFS集群从节点,每一个block都可以在多个datanode上存储多个副本(副本数量也可以通过参数设置dfs.replication)
    补充:同一个block不会存储多份(大于1)在同一个datanode上,因为这样没有意义。

    (5)HDFS是设计成适应一次写入,多次读出的场景,且不支持文件的修改

    (注:适合用来做数据分析,并不适合用来做网盘应用,因为,不便修改,延迟大,网络开销大,成本太高)


    HDFS基本操作篇

    2.1HDFS的shell(命令行客户端)操作

    2.1.1 HDFS命令行客户端使用

    HDFS提供shell命令行客户端,使用方法如下:
    在这里插入图片描述

    2.2 命令行客户端支持的命令参数

    
            [-appendToFile <localsrc> ... <dst>]
            [-cat [-ignoreCrc] <src> ...]
            [-checksum <src> ...]
            [-chgrp [-R] GROUP PATH...]
            [-chmod [-R] <MODE[,MODE]... | OCTALMODE> PATH...]
            [-chown [-R] [OWNER][:[GROUP]] PATH...]
            [-copyFromLocal [-f] [-p] <localsrc> ... <dst>]
            [-copyToLocal [-p] [-ignoreCrc] [-crc] <src> ... <localdst>]
            [-count [-q] <path> ...]
            [-cp [-f] [-p] <src> ... <dst>]
            [-createSnapshot <snapshotDir> [<snapshotName>]]
            [-deleteSnapshot <snapshotDir> <snapshotName>]
            [-df [-h] [<path> ...]]
            [-du [-s] [-h] <path> ...]
            [-expunge]
            [-get [-p] [-ignoreCrc] [-crc] <src> ... <localdst>]
            [-getfacl [-R] <path>]
            [-getmerge [-nl] <src> <localdst>]
            [-help [cmd ...]]
            [-ls [-d] [-h] [-R] [<path> ...]]
            [-mkdir [-p] <path> ...]
            [-moveFromLocal <localsrc> ... <dst>]
            [-moveToLocal <src> <localdst>]
            [-mv <src> ... <dst>]
            [-put [-f] [-p] <localsrc> ... <dst>]
            [-renameSnapshot <snapshotDir> <oldName> <newName>]
            [-rm [-f] [-r|-R] [-skipTrash] <src> ...]
            [-rmdir [--ignore-fail-on-non-empty] <dir> ...]
            [-setfacl [-R] [{-b|-k} {-m|-x <acl_spec>} <path>]|[--set <acl_spec> <path>]]
            [-setrep [-R] [-w] <rep> <path> ...]
            [-stat [format] <path> ...]
            [-tail [-f] <file>]
            [-test -[defsz] <path>]
            [-text [-ignoreCrc] <src> ...]
            [-touchz <path> ...]
            [-usage [cmd ...]]
    
    

    2.3 常用命令参数介绍

    -help
    功能:输出这个命令参数手册

    -ls
    功能:显示目录信息
    示例: hadoop fs -ls hdfs://hadoop-server01:9000/
    备注:这些参数中,所有的hdfs路径都可以简写
    –>hadoop fs -ls / 等同于上一条命令的效果

    ==-mkdir ==
    功能:在hdfs上创建目录
    示例:hadoop fs -mkdir -p /aaa/bbb/cc/dd

    -moveFromLocal
    功能:从本地剪切粘贴到hdfs
    示例:hadoop fs - moveFromLocal /home/hadoop/a.txt /aaa/bbb/cc/dd

    -moveToLocal
    功能:从hdfs剪切粘贴到本地
    示例:hadoop fs - moveToLocal /aaa/bbb/cc/dd /home/hadoop/a.txt

    –appendToFile
    功能:追加一个文件到已经存在的文件末尾
    示例:hadoop fs -appendToFile ./hello.txt hdfs://hadoop-server01:9000/hello.txt
    可以简写为:
    Hadoop fs -appendToFile ./hello.txt /hello.txt

    -cat
    功能:显示文件内容
    示例:hadoop fs -cat /hello.txt

    -tail
    功能:显示一个文件的末尾
    示例:hadoop fs -tail /weblog/access_log.1

    -text
    功能:以字符形式打印一个文件的内容
    示例:hadoop fs -text /weblog/access_log.1

    -chgrp
    -chmod
    -chown

    功能:这三个命令跟linux文件系统中的用法一样,对文件所属权限
    示例:
    hadoop fs -chmod 666 /hello.txt
    hadoop fs -chown someuser:somegrp /hello.txt

    -copyFromLocal
    功能:从本地文件系统中拷贝文件到hdfs路径去
    示例:hadoop fs -copyFromLocal ./jdk.tar.gz /aaa/

    -copyToLocal
    功能:从hdfs拷贝到本地
    示例:hadoop fs -copyToLocal /aaa/jdk.tar.gz

    -cp
    功能:从hdfs的一个路径拷贝hdfs的另一个路径
    示例: hadoop fs -cp /aaa/jdk.tar.gz /bbb/jdk.tar.gz.2

    -mv
    功能:在hdfs目录中移动文件
    示例: hadoop fs -mv /aaa/jdk.tar.gz /

    -get
    功能:等同于copyToLocal,就是从hdfs下载文件到本地
    示例:hadoop fs -get /aaa/jdk.tar.gz

    -getmerge
    功能:合并下载多个文件
    示例:比如hdfs的目录 /aaa/下有多个文件:log.1, log.2,log.3,…
    hadoop fs -getmerge /aaa/log.* ./log.sum

    -put
    功能:等同于copyFromLocal
    示例:hadoop fs -put /aaa/jdk.tar.gz /bbb/jdk.tar.gz.2

    -rm
    功能:删除文件或文件夹
    示例:hadoop fs -rm -r /aaa/bbb/

    -rmdir
    功能:删除空目录
    示例:hadoop fs -rmdir /aaa/bbb/ccc

    -df
    功能:统计文件系统的可用空间信息
    示例:hadoop fs -df -h /

    -du
    功能:统计文件夹的大小信息
    示例:
    hadoop fs -du -s -h /aaa/*

    -count
    功能:统计一个指定目录下的文件节点数量
    示例:hadoop fs -count /aaa/

    -setrep
    功能:设置hdfs中文件的副本数量
    示例:hadoop fs -setrep 3 /aaa/jdk.tar.gz

    补充: hadoop dfsadmin -report 用这个命令可以快速定位出哪些节点down掉了,HDFS的容量以及使用了多少,以及每个节点的硬盘使用情况。


    HDFS原理篇

    hdfs的工作机制

    (工作机制的学习主要是为加深对分布式系统的理解,以及增强遇到各种问题时的分析解决能力,形成一定的集群运维能力)

    注:很多不是真正理解hadoop技术体系的人会常常觉得HDFS可用于网盘类应用,但实际并非如此。要想将技术准确用在恰当的地方,必须对技术有深刻的理解

    3.1 概述

    1. HDFS集群分为两大角色:NameNode、DataNode (Secondary Namenode)
    2. NameNode负责管理整个文件系统的元数据(整个hdfs文件系统的目录树和每个文件的block信息)
    3. DataNode 负责管理用户的文件数据块
    4. 文件会按照固定的大小(blocksize)切成若干块后分布式存储在若干台datanode上
    5. 每一个文件块可以有多个副本,并存放在不同的datanode上
    6. Datanode会定期向Namenode汇报自身所保存的文件block信息,而namenode则会负责保持文件的副本数量
    7. HDFS的内部工作机制对客户端保持透明,客户端请求访问HDFS都是通过向namenode申请来进行

    3.2 HDFS写数据流程

    3.2.1 概述

    客户端要向HDFS写数据,首先要跟namenode通信以确认可以写文件并获得接收文件block的datanode,然后,客户端按顺序将文件逐个block传递给相应datanode,并由接收到block的datanode负责向其他datanode复制block的副本

    3.2.2 详细步骤图

    在这里插入图片描述

    3.2.3 详细步骤解析

    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的服务器。

    3.3. HDFS读数据流程

    3.3.1 概述

    客户端将要读取的文件路径发送给namenode,namenode获取文件的元信息(主要是block的存放位置信息)返回给客户端,客户端根据返回的信息找到相应datanode逐个获取文件的block并在客户端本地进行数据追加合并从而获得整个文件

    3.3.2 详细步骤图

    在这里插入图片描述

    3.3.3 详细步骤解析

    1. 跟namenode通信查询元数据,namenode找到文件块所在的datanode服务器
    2. 挑选一台datanode(就近原则,然后随机)服务器,请求建立socket流
    3. datanode开始发送数据(从磁盘里面读取数据放入流,以packet为单位来做校验)
    4. 客户端以packet为单位接收,现在本地缓存,然后写入目标文件

    4 NAMENODE工作机制

    学习目标:理解namenode的工作机制尤其是元数据管理机制,以增强对HDFS工作原理的理解,及培养hadoop集群运营中“性能调优”、“namenode”故障问题的分析解决能力

    问题场景:

    1. 集群启动后,可以查看文件,但是上传文件时报错,打开web页面可看到namenode正处于safemode状态,怎么处理?
    2. Namenode服务器的磁盘故障导致namenode宕机,如何挽救集群及数据?
    3. Namenode是否可以有多个?namenode内存要配置多大?namenode跟集群数据存储能力有关系吗?
    4. 文件的blocksize究竟调大好还是调小好?
      ……

    诸如此类问题的回答,都需要基于对namenode自身的工作原理的深刻理解

    4.1 NAMENODE职责

    NAMENODE职责:
    负责客户端请求的响应
    元数据的管理(查询,修改)

    4.2 元数据管理

    namenode对数据的管理采用了三种存储形式:

    • 内存元数据(NameSystem)
    • 磁盘元数据镜像文件(fsimage)
    • 数据操作日志文件(edits可通过日志运算出元数据)

    4.2.1 元数据存储机制(元数据是对象,有特定的数据结构,可以理解为hashmap结构)

    A、内存中有一份完整的元数据(内存meta data)
    B、磁盘有一个“准完整”的元数据镜像(fsimage)文件(在namenode的工作目录中)
    C、用于衔接内存metadata和持久化元数据镜像fsimage之间的操作日志(edits文件)注:当客户端对hdfs中的文件进行新增或者修改操作,操作记录首先被记入edits日志文件中,当客户端操作成功后,相应的元数据会更新到内存meta.data中

    补充:
    1、fsimage文件是线性结构,都是0和1,很难查找或者修改某条数据,所以才会定期checkpoint。
    2、edits记录的是操作步骤,类似于mysql的binlog
    3、fsimage记录的是这个文件备份了几份,分别叫什么名称
    在这里插入图片描述
    4、secondary namenode建议不和namenode在一个节点启动,因为它会拷贝元数据,加载到内存生成fsimage,会占用namenode的内存。(最简版)
    5、在hadoop的高可用机制+Federation机制中,没有SecondaryNamenode,可以通过启动SecondaryNamenode进行验证,会报一个错误:“它的功能被StandbyNamenode取代”。(在启动的那台机器的logs文件夹里面的SecondaryNamenode.log)。(完全版)

    4.2.2 元数据手动查看

    可以通过hdfs的一个工具来查看edits中的信息
    bin/hdfs oev -i edits -o edits.xml
    bin/hdfs oiv -i fsimage_0000000000000000087 -p XML -o fsimage.xml

    4.2.3 元数据的checkpoint

    每隔一段时间,会由secondary namenode将namenode上最新的edits(下载过的namenode会删除)和fsimage(第一次时会下载fsimage,以后不会)下载到secondary namenode中,并加载到内存进行merge(这个过程称为checkpoint)
    checkpoint的详细过程
    在这里插入图片描述

    checkpoint操作的触发条件配置参数

    dfs.namenode.checkpoint.check.period=60 #检查触发条件是否满足的频率,60秒
    dfs.namenode.checkpoint.dir=file://KaTeX parse error: Expected 'EOF', got '#' at position 36: …/namesecondary #̲以上两个参数做checkpoi…{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的元数据。

    4.2.4 元数据目录说明

    在第一次部署好Hadoop集群的时候,我们需要在NameNode(NN)节点上格式化磁盘:

    $HADOOP_HOME/bin/hdfs namenode -format

    格式化完成之后,将会在$ dfs. namenode .name.dir/current目录下如下的文件结构

    current/
    |-- VERSION
    |-- edits_*
    |-- fsimage_0000000000008547077
    |-- fsimage_0000000000008547077.md5
    `-- seen_txid

    其中的dfs.name.dir是在hdfs-site.xml文件中配置的,默认值如下:

    <property>
      <name>dfs.name.dir</name>
      <value>file://${hadoop.tmp.dir}/dfs/name</value>
    </property>
    

    hadoop.tmp.dir是在core-site.xml中配置的,默认值如下

    <property>
      <name>hadoop.tmp.dir</name>
      <value>/tmp/hadoop-${user.name}</value>
      <description>A base for other temporary directories.</description>
    </property>
    

    dfs. namenode.name.dir属性可以配置多个目录,
    如/data1/dfs/name,/data2/dfs/name,/data3/dfs/name,…。各个目录存储的文件结构和内容都完全一样,相当于备份,这样做的好处是当其中一个目录损坏了,也不会影响到Hadoop的元数据,特别是当其中一个目录是NFS(网络文件系统Network File System,NFS)之上,即使你这台机器损坏了,元数据也得到保存。
    下面对$dfs. namenode .name.dir/current/目录下的文件进行解释。

    1. VERSION文件是Java属性文件,内容大致如下:

    #Fri Nov 15 19:47:46 CST 2013
    namespaceID=934548976
    clusterID=CID-cdff7d73-93cd-4783-9399-0a22e6dce196
    cTime=0
    storageType=NAME_NODE
    blockpoolID=BP-893790215-192.168.24.72-1383809616115
    layoutVersion=-47

    其中
      (1)、namespaceID是文件系统的唯一标识符,在文件系统首次格式化之后生成的;
      (2)、storageType说明这个文件存储的是什么进程的数据结构信息(如果是DataNode,storageType=DATA_NODE);
      (3)、cTime表示NameNode存储时间的创建时间,由于我的NameNode没有更新过,所以这里的记录值为0,以后对NameNode升级之后,cTime将会记录更新时间戳;
      (4)、layoutVersion表示HDFS永久性数据结构的版本信息, 只要数据结构变更,版本号也要递减,此时的HDFS也需要升级,否则磁盘仍旧是使用旧版本的数据结构,这会导致新版本的NameNode无法使用;
      (5)、clusterID是系统生成或手动指定的集群ID,在-clusterid选项中可以使用它;如下说明

    a、使用如下命令格式化一个Namenode:

    $HADOOP_HOME/bin/hdfs namenode -format [-clusterId <cluster_id>]

    选择一个唯一的cluster_id,并且这个cluster_id不能与环境中其他集群有冲突。如果没有提供cluster_id,则会自动生成一个唯一的ClusterID。
    b、使用如下命令格式化其他Namenode:

    $HADOOP_HOME/bin/hdfs namenode -format -clusterId <cluster_id>

    c、升级集群至最新版本。在升级过程中需要提供一个ClusterID,例如:

    $ HADOOP_PREFIX_HOME/bin/hdfs start namenode --config
    $ HADOOP_CONF_DIR -upgrade -clusterId <cluster_ID>

    如果没有提供ClusterID,则会自动生成一个ClusterID。
      (6)、blockpoolID:是针对每一个Namespace所对应的blockpool的ID,上面的这个BP-893790215-192.168.24.72-1383809616115就是在我的ns1的namespace下的存储块池的ID,这个ID包括了其对应的NameNode节点的ip地址。
      
    2. $dfs.namenode.name.dir/current/seen_txid非常重要,是存放transactionId的文件,format之后是0,它代表的是namenode里面的edits_*文件的尾数,namenode重启的时候,会按照seen_txid的数字,循序从头跑edits_0000001~到seen_txid的数字。所以当你的hdfs发生异常重启的时候,一定要比对seen_txid内的数字是不是你edits最后的尾数,不然会发生建置namenode时metaData的资料有缺少,导致误删Datanode上多余Block的资讯。

    1. $dfs.namenode.name.dir/current目录下在format的同时也会生成fsimage和edits文件,及其对应的md5校验文件。
      补充:seen_txid
      文件中记录的是edits滚动的序号,每次重启namenode时,namenode就知道要将哪些edits进行加载edits

    5 DATANODE的工作机制

    问题场景:
    1、集群容量不够,怎么扩容?
    2、如果有一些datanode宕机,该怎么办?
    3、datanode明明已启动,但是集群中的可用datanode列表中就是没有,怎么办?

    以上这类问题的解答,有赖于对datanode工作机制的深刻理解

    5.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>
    

    5.2 观察验证DATANODE功能

    上传一个文件,观察文件的block具体的物理存放情况:

    在每一台datanode机器上的这个目录中能找到文件的切块:
    /home/hadoop/app/hadoop-2.4.1/tmp/dfs/data/current/BP-193442119-192.168.2.120-1432457733977/current/finalized

    5.3元数据目录(自己添加,实测有效)

    其中的dfs.data.dir是在hdfs-site.xml文件中配置的,默认值如下:

    <property>
      <name>dfs.data.dir</name>
      <value>file://${hadoop.tmp.dir}/dfs/name</value>
    </property>
    

    dfs. datanode data.dir属性可以配置多个目录,
    如/data1/dfs/ data,/data2/dfs/ data,/data3/dfs/ data,…。datanode配置多块磁盘后,会将这些磁盘统一看成它的空间。并发时有优势,可以往不同的磁盘写数据,磁盘可以并行。相当于扩容。

    补充:block块默认128M,最小配置为1M


    HDFS应用开发篇

    6. HDFS的java操作

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

    6.1 搭建开发环境

    1. 引入依赖
    <dependency>
        <groupId>org.apache.hadoop</groupId>
        <artifactId>hadoop-client</artifactId>
        <version>2.6.1</version>
    </dependency>
    

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

    1. window下开发的说明
      建议在linux下进行hadoop应用的开发,不会存在兼容性问题。如在window上做客户端应用开发,需要设置以下环境:
      A、在windows的某个目录下解压一个hadoop的安装包
      B、将安装包下的lib和bin目录用对应windows版本平台编译的本地库替换
      C、在window系统中配置HADOOP_HOME指向你解压的安装包
      D、在windows系统的path变量中加入hadoop的bin目录

    6.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的实例,而是一个本地文件系统的客户端对象

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

    在这里插入图片描述

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

    6.4.1 文件的增删改查

    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.xml
    		Configuration conf = new Configuration();
    		conf.set("fs.defaultFS", "hdfs://hdp-node01: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://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());
    		}
    	}
    }
    

    6.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");
    
    	}
    	
    		/**
    	 * 通过流的方式上传文件到hdfs
    	 * @throws Exception
    	 */
    	@Test
    	public void testUpload() throws Exception {
    		
    		FSDataOutputStream outputStream = fs.create(new Path("/angelababy.love"), true);
    		FileInputStream inputStream = new FileInputStream("c:/angelababy.love");
    		
    		IOUtils.copy(inputStream, outputStream);
    		
    	}
    	
    	@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);
    	}
    }
    

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

    7. 案例1:开发shell采集脚本

    7.1需求说明

    点击流日志每天都10T,在业务应用服务器上,需要准实时上传至数据仓库(Hadoop HDFS)上

    7.2需求分析

    一般上传文件都是在凌晨24点操作,由于很多种类的业务数据都要在晚上进行传输,为了减轻服务器的压力,避开高峰期。
    如果需要伪实时的上传,则采用定时上传的方式

    7.3技术分析

    HDFS SHELL: hadoop fs –put xxxx.tar /data 还可以使用 Java Api
    满足上传一个文件,不能满足定时、周期性传入。
    定时调度器:
    Linux crontab
    crontab -e
    */5 * * * * $home/bin/command.sh //五分钟执行一次
    系统会自动执行脚本,每5分钟一次,执行时判断文件是否符合上传规则,符合则上传

    7.4实现流程

    7.4.1日志产生程序

    日志产生程序将日志生成后,产生一个一个的文件,使用滚动模式创建文件名。
    在这里插入图片描述
    日志生成的逻辑由业务系统决定,比如在log4j配置文件中配置生成规则,如:当xxxx.log 等于10G时,滚动生成新日志

    log4j.logger.msg=info,msg
    log4j.appender.msg=cn.maoxiangyi.MyRollingFileAppender
    log4j.appender.msg.layout=org.apache.log4j.PatternLayout
    log4j.appender.msg.layout.ConversionPattern=%m%n
    log4j.appender.msg.datePattern='.'yyyy-MM-dd
    log4j.appender.msg.Threshold=info
    log4j.appender.msg.append=true
    log4j.appender.msg.encoding=UTF-8
    log4j.appender.msg.MaxBackupIndex=100
    log4j.appender.msg.MaxFileSize=10GB
    log4j.appender.msg.File=/home/hadoop/logs/log/access.log
    

    细节:
    1、如果日志文件后缀是1\2\3等数字,该文件满足需求可以上传的话。把该文件移动到准备上传的工作区间。
    2、工作区间有文件之后,可以使用hadoop put命令将文件上传。

    阶段问题:
    1、待上传文件的工作区间的文件,在上传完成之后,是否需要删除掉。

    7.4.2伪代码

    使用ls命令读取指定路径下的所有文件信息,
    ls | while read line
    //判断line这个文件名称是否符合规则
    if line=access.log.* (
    将文件移动到待上传的工作区间
    )

    //批量上传工作区间的文件
    hadoop fs –put xxx

    脚本写完之后,配置linux定时任务,每5分钟运行一次。

    7.5代码实现

    代码第一版本,实现基本的上传功能和定时调度功能
    在这里插入图片描述

    代码第二版本:增强版V2(基本能用,还是不够健全)
    在这里插入图片描述
    在这里插入图片描述

    7.6效果展示及操作步骤

    1、日志收集文件收集数据,并将数据保存起来,效果如下:
    在这里插入图片描述
    2、上传程序通过crontab定时调度
    在这里插入图片描述
    3、程序运行时产生的临时文件
    在这里插入图片描述
    4、Hadoo hdfs上的效果
    在这里插入图片描述

    8. 案例2:开发JAVA采集程序

    8.1 需求

    从外部购买数据,数据提供方会实时将数据推送到6台FTP服务器上,我方部署6台接口采集机来对接采集数据,并上传到HDFS中

    提供商在FTP上生成数据的规则是以小时为单位建立文件夹(2016-03-11-10),每分钟生成一个文件(00.dat,01.data,02.dat,…)

    提供方不提供数据备份,推送到FTP服务器的数据如果丢失,不再重新提供,且FTP服务器磁盘空间有限,最多存储最近10小时内的数据

    由于每一个文件比较小,只有150M左右,因此,我方在上传到HDFS过程中,需要将15分钟时段的数据合并成一个文件上传到HDFS

    为了区分数据丢失的责任,我方在下载数据时最好进行校验

    8.2 设计分析

    在这里插入图片描述

    划重点(∩_∩)



    本人程序媛一枚,因为离港澳较近,周末兼职港澳人肉代购。

    欢迎各位大佬添加本人微信,还会经常有点赞活动送价值不菲的小礼品哦。

    即使现在不需要代购,等以后有了女(男)朋友、有了宝宝就肯定会需要的喽。

    动动手指头,扫码一下,就当是对本博文的支持嘛,也是对一个平凡、勤劳、勇敢、秀外慧中等等优点的程序媛莫大的支持哈。
    在这里插入图片描述

    展开全文
  • 1.HDFS原理与架构说明 2.HDFS的NameNode和DataNode 3.HDFS缓存机制(Cache) 4.HDFS快照(Snapshot) 5.HDFS命令行操作 6.HDFS的Java API编程 7.HDFS Web HDFS API编程实践 8.HDFS企业级项目实战:某企业...
  • HDFS和YARN HA部署

    2019-04-17 11:20:04
    任意时刻只能有一个nn(active状态)去写,nn standby 读 jn部署台数是奇数2n+1,active nn把编辑日志写道jn,要求至少 n/2+1台jn是好的。 如果有3台jn,三台都是好的,那么会写三台,如果有一台是坏的,那么满足3+1...
  • HDFS&YARN HA部署

    2019-04-10 18:34:06
    HDFS&YARN HA部署 一:集群资源 集群资源:3台4核4G的服务器(hadoop001,hadoop002,hadoop003),均为阿里云服务器 需要的安装包: ​ 1.jdk压缩包(jdk-8u45-linux-x64.gz) ​ 2.Hadoop压缩包(hadoop-2.6.0-cdh...
  • 撸了今年阿里、头条和美团的面试,我有一个重要发现.......>>> ...
  • 一、JDK部署 在hadoop001机器下可以把app目录下的软件都传输到hadoop002目录下,测试,速度贼快。 [hadoop@hadoop001 app]$ scp * hadoop002:/home/hadoop/app jdk部署在 /usr/java 目录下 ...
  • HDFS简介及其功能

    万次阅读 多人点赞 2018-01-19 11:03:48
    HDFS教程在这个HDFS教程博客中继续前进之前,让我带您介绍一些与HDFS相关的疯狂统计信息:Facebook在2010年声称拥有存储21 PB数据的最大HDFS集群之一。在2012年,Facebook宣布他们拥有最大的单个HDFS集群,数据量...
  • 大数据(二) - HDFS

    千次阅读 2016-02-25 05:35:23
    HDFS优点  高容错性  数据自动保存多个副本  副本丢失后,自动恢复  适合批处理  移动计算而非数据  数据位置暴露给计算框架  适合大数据处理  GB、TB、甚至PB级数据  百万规模以上的文件数量
  • 一、HDFS介绍 上篇文章已经讲到了,随着数据量越来越大,在一台机器上已经无法存储所有的数据了,那我们会将这些数据分配到不同的机器来进行存储,但是这就带来一个问题:不方便管理和维护 所以,我们就希望有一个...
  • HDFS文件夹上传和下载等命令

    万次阅读 2019-05-26 16:44:22
    一、用浏览器查看hdfs目录下的文件夹和文件 二、HDFS命令 零、码仙励志 白茶清欢无别事,我在等风也等你 一、用浏览器查看hdfs目录下的文件夹和文件 在浏览器中输入IP地址+端口号50070就可以,如下: 然后...
  • hdfs统计行数和统计文件大小

    万次阅读 2014-12-10 10:49:47
    hdfs统计文件行数和文件大小命令
  • hdfs查看各级目录文件的大小

    万次阅读 2017-10-24 10:28:26
    1.hadoop dfs -du -h 查看各级文件大小 会出现两列数字,第一列为单个文件实际大小,第二列为备份大小。 2.hadoop dfs -ls -h也可以查看文件大小 但是当子文件目录过多时 统计的大小为0
  • 1、HDFS上创建目录 ${HADOOP_HOME}/bin/hdfs dfs -mkdir /demo1 2、上传本地文件到HDFS上 ${HADOOP_HOME}/bin/hdfs dfs -put ${HADOOP_HOME}/etc/hadoop/core-site.xml /demo1 3、上传本地文件到HDFS上 ${...
  • 查看HDFS文件系统数据的三种方法

    万次阅读 2017-01-04 11:10:56
    1、使用插件——Hadoop-Eclipse-Plugin 2、HDFS Web界面 3、shell命令
  • Linux上传本地文件到HDFS

    万次阅读 2018-01-15 16:15:20
    上一篇文章记录了如何使用java代码将本地文件上传到HDFS中,之后我又尝试了直接使用Linux命令上传文件,过程如下。 su hdfs //切换到HDFS用户 cd /home/hdfs/files //进入到服务器中文件所在目录 ls //查看此目录...
  • 查看hdfs文件夹大小

    万次阅读 2018-06-19 16:26:11
    例如查看ods库大小:hadoop fs -du -s -h /user/hive/warehouse/ods.db如果要查看其他文件夹则改成该文件夹路径。去掉-h选项则按Byte单位显示。
  • 使用命令查看hdfs的状态

    万次阅读 2017-03-21 20:13:56
    使用如下命令 : hdfs dfsadmin -report
1 2 3 4 5 ... 20
收藏数 235,766
精华内容 94,306
关键字:

hdfs