大数据入门第六天——HDFS详解

一、概述

  1.HDFS中的角色

    Block数据:

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

  1. 基本存储单位,一般大小为64M(配置大的块主要是因为:1)减少搜寻时间,一般硬盘传输速率比寻道时间要快,大的块可以减少寻道时间;2)减少管理块的数据开销,每个块都需要在NameNode上有对应的记录;3)对数据块进行读写,减少建立网络的连接成本)

  2. 一个大文件会被拆分成一个个的块,然后存储于不同的机器。如果一个文件少于Block大小,那么实际占用的空间为其文件的大小

  3. 基本的读写S#x5355;位,类似于磁盘的页,每次都是读写一个块

  4. 每个块都会被复制到多台机器,默认复制3份

      NameNode:

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

    Secondary NameNode:

      定时与NameNode进行同步(定期合并文件系统镜像和编辑日&#x#x5FD7;,然后把合并后的传给NameNode,替换其镜像,并清空编辑日志,类似于CheckPoint机制),但NameNode失效后仍需要手工将其设置成主机——namenode的冷备份

      关于这点,可以参考http://blog.csdn.net/scgaliguodong123_/article/details/46335427

    DataNode:

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

     文件会按照固定的大小(blocksize)切成若干块(由上传的客户端进行切块处理,这样不大于128M切块大小的实际是多少就是多少)后分布式存储在若干台datanode上

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

  详细角色信息,参考:https://www.w3cschool.cn/hadoop/xvmi1hd6.html

  漫画式的讲解,参考:https://www.cnblogs.com/raphael5200/p/5497218.html

二、读写数据流程

  简要的说明参考上文漫画式讲解处

  专业深入讲解,参考https://www.cnblogs.com/codeOfLife/p/5375120.html

  1.写数据流程

、根namenode通信请求上传文件,namenode检查目标文件是否已存在,父目录是否存在
、namenode返回是否可以上传
、client请求第一个 block该传输到哪些datanode服务器上
、namenode返回3个datanode服务器ABC
、client请求3台dn中的一台A上传数据(本质上是一个RPC调用,建立pipeline),A收到请求会继续调用B,然后B调用C,将真个pipeline建立完成,逐级返回客户端
、client开始往A上传第一个block(先从磁盘读取数据放到一个本地内存缓存),以packet为单位,A收到一个packet就会传给B,B传给C;A每传一个packet会放入一个应答队列等待应答
、当一个block传输完成之后,client再次请求namenode上传第二个block的服务器。

  图解请参考上文

  注意:HDFS只允许修改文件名/文件追加等,无法直接修改原来的文件!

  2.读数据流程

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

三、HDFS元数据管理与解析

  1.元数据管理原理

    元数据分类:

  • 第一类是文件和目录自身的属性信息,例如文件名、目录名、父目录信息、文件大小、创建时间、修改时间等。
  • 第二类记录文件内容存储相关信息,例如文件块情况、副本个数、每个副本所在的Data Node 信息等。
  • 第三类用来记录HDFS中所有Data Node信息,用于Data Node管理。

    存储机制:

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

      B、磁盘有一个“准完整”的元数据镜像(fsimage)文件(在namenode的工作目录中),整个运行过程中,fsimage是不会改变的!editslog的更新会同步到内存

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

    更多参考http://blog.csdn.net/xiaming564/article/details/23165253

        http://blog.csdn.net/chenkfkevin/article/details/61196409

    这里就能很清楚的知道secondary namenode的工作原理了!

    关于元数据的存取流程与分析,参考网友的白话讲解http://blog.csdn.net/lepton126/article/details/53183037

    2.修改工作目录

    之前我们配置过hadoop.tmp.dir来设置临时目录,这里可以通过以下参数设置HDFS工作目录

配置文件:hdfs-site.xml
参数名:dfs.namenode.name.dir
格式:file://${hadoop.tmp.dir}/dfs/name
说明:Determines where on the local filesystem the DFS name node should store the name table(fsimage).
If this is a comma-delimited list of directories then the name table is replicated in all of the directories, for redundancy.
示例:
<property>
<name>dfs.namenode.name.dir</name>
<value>/home/hadoop/name1,/home/hadoop/name2</value>
</property>

  //当然,dfs.datanode.data.dir也是可以配置的,这样重新格式化后就可以重新使用工作目录了!点击查看对比

  注:如果还在使用dfs.name.dir/dfs.data.dir,请查看官网配置的deprecated properties

    3.元数据各目录解析

      详细解析,参考https://www.iteblog.com/archives/967.html

              http://blog.csdn.net/opensure/article/details/51452058

四、HDFS的Java-API操作

   完整API参考官网:http://hadoop.apache.org/docs/current/api/

  1.基本的增删改查API:

package com.hdfs.demo;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.*;
import org.junit.Before;
import org.junit.Test; import java.net.URI; /**
* HDFS客户端demo
*
* @author zcc ON 2018/1/28
**/
public class HdfsClientDemo {
FileSystem fs = null;
@Before
public void init() throws Exception{
// 如配置完环境变量,下行可省略
// System.setProperty("hadoop.home.dir", "F:\\work\\hadoop-2.6.4");
Configuration conf = new Configuration();
// 配置文件系统(注意hosts的配置)
// conf.set("fs.defaultFS","hdfs://mini1:9000");
// 拿到一个操作的客户端实例对象(此处使用3个参数则上一行省略)
fs = FileSystem.get(new URI("hdfs://mini1:9000"),conf,"hadoop");
}
@Test
public void testUpload() throws Exception{
// 就对应get的别名
fs.copyFromLocalFile(new Path("F:/c.log"),new Path("/c.log.copy"));
// 关闭
fs.close();
}
@Test
public void testDelete() throws Exception{
// 第一个是Path,第二个为是否递归删除
boolean b = fs.delete(new Path("/c.log.copy"), true);
System.out.println("删除状态:" + b);
fs.close();
}
@Test
public void testList() throws Exception{
/*
一般而言,大数据方面使用迭代器场景居多,因为Iterator它本身并不存数据(可以查看源码)
它只是提供了几个简单的方法帮你去取数据,而使用ArrayList则是直接把数据拿过来了,大数据
量下不适合
*/
RemoteIterator<LocatedFileStatus> iterator = fs.listFiles(new Path("/"), true);
while (iterator.hasNext()) {
LocatedFileStatus next = iterator.next();
System.out.println("Path:" + next.getPath());
System.out.println("BlockSize" + next.getBlockSize());
System.out.println("Name:" + next.getPath().getName());
} // Path:hdfs://mini1:9000/1.txt
// BlockSize134217728
// Name:1.txt
fs.close();
} /**
* 既可以遍历文件,又可以遍历文件夹
* @throws Exception
*/
@Test
public void testList2() throws Exception{
FileStatus[] files = fs.listStatus(new Path("/"));
for (FileStatus file : files) {
System.out.println(file.getPath().getName());
if (file.isFile()) {
System.out.println("it is a file");
}
}
}
}

  这里引入Java中迭代器的机制浅谈,供参考https://www.cnblogs.com/hasse/p/5024193.html

  更多实例,参考:http://blog.csdn.net/litianxiang_kaola/article/details/70983904

  2.流操作API

    相对那些封装好的方法而言的更底层一些的操作方式上层那些mapreduce   spark等运算框架,去hdfs中获取数据的时候,就是调的这种底层的api

package com.hdfs.stream;

import org.apache.commons.io.IOUtils;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.*;
import org.junit.Before;
import org.junit.Test; import java.io.FileInputStream;
import java.io.FileOutputStream;
import java.io.IOException;
import java.net.URI; /**
* HDFS的流API操作
*
* @author zcc ON 2018/1/30
**/
public class HdfsStreamAccess {
FileSystem fs = null;
@Before
public void init() throws Exception{
Configuration conf = new Configuration();
fs = FileSystem.get(new URI("hdfs://mini1:9000"),conf,"hadoop");
} /**
* 通过流的操作,就可以为上层MR程序提供服务,例如跑任务只需要跑60M,这样就不用通过
* 之前的直接得到一整个文件的复杂处理
* 而建立文件夹等直接通过简单API即可!
* @throws Exception
*/
@Test
public void testUpload() throws Exception{
// 输出到HDFS
FSDataOutputStream outputStream = fs.create(new Path("/angle"), true);
// 本地输入流
FileInputStream inputStream = new FileInputStream("F:\\c.log");
// 通过IOUtils进行拷贝(使用更加通用的IOUtils)
IOUtils.copy(inputStream, outputStream);
}
@Test
public void testDownload() throws Exception{
FSDataInputStream inputStream = fs.open(new Path("/angle"));
FileOutputStream outputStream = new FileOutputStream("F:\\c-download.log");
IOUtils.copy(inputStream, outputStream);
} /**
* 指定随机长度读取
*/
@Test
public void testRandomAccess() throws Exception{
FSDataInputStream inputStream = fs.open(new Path("/angle"));
inputStream.seek(12);
FileOutputStream outputStream = new FileOutputStream("F:\\c-random.log");
// 从12字节读到末尾(也可以通过while结合自定义的count等来控制读文件的大小)
// 后续会避免读取读到单词一半这样的问题
// IOUtils.copy(inputStream, outputStream);
IOUtils.copyLarge(inputStream, outputStream, 12, 100);
} /**
* 以下模拟实现:获取一个文件的所有block位置信息,然后读取指定block中的内容
* @throws IllegalArgumentException
* @throws IOException
*/
@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();
}
}

五、案例:shell脚本日志采集

  采用shell编写的脚本如下:

#!/bin/bash

#set java env
export JAVA_HOME=/home/hadoop/app/jdk1..0_51
export JRE_HOME=${JAVA_HOME}/jre
export CLASSPATH=.:${JAVA_HOME}/lib:${JRE_HOME}/lib
export PATH=${JAVA_HOME}/bin:$PATH #set hadoop env
export HADOOP_HOME=/home/hadoop/app/hadoop-2.6.
export PATH=${HADOOP_HOME}/bin:${HADOOP_HOME}/sbin:$PATH #版本1的问题:
#虽然上传到Hadoop集群上了,但是原始文件还在。如何处理?
#日志文件的名称都是xxxx.log1,再次上传文件时,因为hdfs上已经存在了,会报错。如何处理? #如何解决版本1的问题
# 、先将需要上传的文件移动到待上传目录
# 、在讲文件移动到待上传目录时,将文件按照一定的格式重名名
# /export/software/hadoop.log1 /export/data/click_log/xxxxx_click_log_{date} #日志文件存放的目录
log_src_dir=/home/hadoop/logs/log/ #待上传文件存放的目录
log_toupload_dir=/home/hadoop/logs/toupload/ #日志文件上传到hdfs的根路径
hdfs_root_dir=/data/clickLog// #打印环境变量信息
echo "envs: hadoop_home: $HADOOP_HOME" #读取日志文件的目录,判断是否有需要上传的文件
echo "log_src_dir:"$log_src_dir
ls $log_src_dir | while read fileName
do
if [[ "$fileName" == access.log.* ]]; then
# if [ "access.log" = "$fileName" ];then
date=`date +%Y_%m_%d_%H_%M_%S`
#将文件移动到待上传目录并重命名
#打印信息
echo "moving $log_src_dir$fileName to $log_toupload_dir"xxxxx_click_log_$fileName"$date"
mv $log_src_dir$fileName $log_toupload_dir"xxxxx_click_log_$fileName"$date
#将待上传的文件path写入一个列表文件willDoing
echo $log_toupload_dir"xxxxx_click_log_$fileName"$date >> $log_toupload_dir"willDoing."$date
fi done
#找到列表文件willDoing
ls $log_toupload_dir | grep will |grep -v "_COPY_" | grep -v "_DONE_" | while read line
do
#打印信息
echo "toupload is in file:"$line
#将待上传文件列表willDoing改名为willDoing_COPY_
mv $log_toupload_dir$line $log_toupload_dir$line"_COPY_"
#读列表文件willDoing_COPY_的内容(一个一个的待上传文件名) ,此处的line 就是列表中的一个待上传文件的path
cat $log_toupload_dir$line"_COPY_" |while read line
do
#打印信息
echo "puting...$line to hdfs path.....$hdfs_root_dir"
hadoop fs -put $line $hdfs_root_dir
done
mv $log_toupload_dir$line"_COPY_" $log_toupload_dir$line"_DONE_"
done

脚本文件

  加入定时任务调度,参考大数据之Linux基础

上一篇:前后端分离开发——模拟数据mock.js


下一篇:Power-BI 主要城市商品房销售分析