2. hdfs

一.Hdfs的shell

所有hadoop的fs的shell均用uri路径作为参数

uri格式:schema://authority/path.hdfs的schema是hdfs.其中,schema和authority是可选的,若未指定,hdfs命令使用配置文件中的authority(namenode:namenodePort)

superuser

启动进程的用户是superuser,拥有最高权限,可以做任何事情

dfsadmin命令使用:

#-setQuota:限制该目录下所存文件的总大小限制,改大小算上副本的大小(1GB文件要分配3GBquota)
bin/hdfs dfsadmin -setQuota 10 lisi

#删除该目录下的quota限制
bin/hdfs dfsadmin -clrSpaceQuota  lisi

#dfs -count:计算目录下的文件夹个数,文件个数,所占大小(非quota,不算副本大小)
bin/hdfs dfs -count -q -h /lisi
#The output columns with -count -q are: QUOTA, REMAINING_QUATA, SPACE_QUOTA, REMAINING_SPACE_QUOTA, DIR_COUNT, FILE_COUNT, CONTENT_SIZE, FILE_NAME

#The -h option shows sizes in human readable format.

dfs文件操作

#查看内容
hdfs dfs -text 文件
#只看一段内容
hdfs dfs -text 文件 | more

#追加文件(只能在结尾追加)
hdfs dfs -appendToFile localFile remoteFile

二.NameNode

fsimage：元数据镜像文件。存储某一时段NameNode内存元数据信息。

保存最近一次checkpoint的时间namenode内存中的文件树镜像

【注】：checkpoint时机的选择，checkpoint使得新的fsimage持久化

（1）fs.checkpoint.size：设置了上次checkpoint后，editslog文件的大小限制

（2）fs.checkpoint.period：上次fsimage保存过一段时间后触发checkpoint

editslog：操作日志文件

记录上次fsimmage持久化后namenode发生的变化

（1）配置在dfs.namenode.edits.dir的文件位置

（2）这些edits文件已经被编码,查看方法如下

#1.启动服务器
bin/hdfs oiv -i 某个fsimage文件

#2.查看内容
bin/hdfs dfs -ls  -R webhdfs://127.0.0.1:5978/

#3.导出结果
bin/hdfs oiv -p XML -i  tmp/dfs/name/current/fsimage_0000000000000000055  -o fsimage.xml

#4.查看edtis内容
bin/hdfs oev -i tmp/dfs/name/current/edits_0000000000000000057-0000000000000000186   -o edits.xml

namenode启动流程

（1）namenode按照fsimage文件的内容顺序读取，构建hdfs文件的目录树。再按照editslog中的操作日志对文件数操作，形成最新的文件树和对应的blockid，此时全部目录树生产完毕，只是缺少每个文件对应的block对应的datanode列表信息，这些信息，需要datanode通过blockreport上报。所以，namenode在fsimage加载完毕后，进入rpc等待时间（safemode），等待所有的datanode汇报blockreport

（2）datanode启动后，会扫描本机$(dfs.data.dir)目录下的block信息，以long数组的形式发送给namenode，namenode解析获得的blockid和datanode的地址，插入到文件树节点中，形成hdfs所有目录树

namenode的HA（主备）

HA要保证active和standby节点信息的一致(事务),edits文件件就是transaction file

三.HDFS的读取，写入操作流程

staging(数据存储)：client端上传文件到hdfs时，会现在本地缓存数据，当数据达到1个block大小时，请求nammenode分配一个block，Namenode把block所在datanode的地址告诉DfsClient，DfsClient直接和Datanode通信，把数据写到Datanode节点的一个block块中

读取过程

（1）首先调用FileSystem对象的open方法，其实是一个DistributedFileSystem的实例

（2）2.DistributedFileSystem通过与namenode的rpc（ClientProtocol）获得文件的第一批个block的locations（包括复本位置），

（3）前两步会返回一个FSDataInputStream对象，该对象会被封装成DFSInputStream对象，DFSInputStream可以方便的管理datanode和namenode数据流。客户端调用read方法，DFSInputStream最会找出离客户端最近的datanode并连接。

（4）数据从datanode源源不断的流向客户端。

（5）当地一个block读取完毕，client会关闭与该块datanode的链接，接着读取下一块

（6）接着DFSInputStream去namenode拿下一批blocks的location，然后继续读，如果所有的块都读完，这时就会关闭掉所有的流

四.小文件的解决方案

har文件

Hadoop Archives的出现是为了缓解大量小文件消耗namenode内存的问题。HAR文件在hdfs上构建了一个层次化的文件系统。hadoop的archive命令通过运行mapreduce任务用来打包小文件成一个har文件。har文件在hdfs中表现为directory目录，可以直接用hdfs dfs -ls查看

【注】：通过har读取一个文件并不必直接从hdfs读取文件高效，因为har文件多了读取索引内部小文件的index。har文件作为map任务输入时，不会以一个文件输入，而是以内部小文件的方式创建map任务

#创建文件
hadoop archive -archiveName 文件.har -p SRC DEST
#查看文件
hadoop fs -lsr har:///DEST/xxx.har

自己控制，把多个小文件write成一个hdfs的文件

public class TestSmallFile {
public static void main(String[] args) throws URISyntaxException, IOException, InterruptedException {
Configuration conf = new Configuration();
URI uri = new URI("hdfs://10.1.198.60:8020");
FileSystem fs = FileSystem.get(uri, conf,"hdfs");

Path path = new Path("/combine.txt");
FSDataOutputStream outputStream = fs.create(path);
for (File f : new File("d://mydir").listFiles()){
BufferedReader reader = new BufferedReader(new InputStreamReader(new FileInputStream(f.getAbsoluteFile())));
String line;
while ((line=reader.readLine())!=null){
outputStream.write(line.getBytes());
}
reader.close();
}
outputStream.close();
}
}

使用sequencefile

（1）sequencefile的结构：sequencefile由header和recod组成，header中记录了此文件的record是否经过压缩，key的classname，value的classname等；record是一个键值对。value部分支持2中压缩：record压缩：对每个record的value部分压缩block压缩：将多个record压缩成一个block（sequencefile内部的block）

（2）小文件解决方案：使用filename作为key，filecontents作为value，把每个文件以的形式写入到sequencefile中。sequencefile支持split，可被压缩（bzip2可被压缩）

public class MyTest {
public static void main(String[] args) throws Exception {
Configuration conf = new Configuration();
URI uri = new URI("hdfs://10.1.198.60:8020");
FileSystem fs = FileSystem.get(uri, conf,"hdfs");
Path path = new Path("/myseqfile.seq");

//通过writer向文档中写入记录
SequenceFile.Writer writer = SequenceFile.createWriter(fs, conf, path,Text.class, Text.class);
writer.append(new Text("filename"),new Text("filevalue"));
IOUtils.closeStream(writer);//关闭write流

//通过reader从文档中读取记录
Text key=new Text();
Text value=new Text();
SequenceFile.Reader reader=new SequenceFile.Reader(fs,path,conf);
while(reader.next(key,value)){
System.out.println(key);
System.out.println(value);
}
IOUtils.closeStream(reader);//关闭read流
}
}

Mapfile

（1）mapfile是key排序后的sequencefile

（2）mapfile由两部分组成：data和index

（3）index中记录了每个Record的key值，以及改Record对应的value在mapfile中的偏移量。访问mapfile时，要先把index部分加载到内存，通过index映射，快速找到record的位置，因此，mapfile的访问速度会很快，单笔sequencefile消耗内存

（4）mapfile不会把所有的key都映射在index中，默认每隔128条记录存储一个映射,io.map.index.interval属性或MapFIle.Writer的setIndexInterval()方法修改key的映射间隔

（5）mapfile的KeyClass要实现WritableComparable接口,即Key值是可比较的。

Configuration conf=new Configuration();
FileSystem fs=FileSystem.get(conf);
Path mapFile=new Path("mapFile.map");

//Writer内部类用于文件的写操作,假设Key和Value都为Text类型
MapFile.Writer writer=new MapFile.Writer(conf,fs,mapFile.toString(),Text.class,Text.class);

//通过writer向文档中写入记录
writer.append(new Text("key"),new Text("value"));
IOUtils.closeStream(writer);//关闭write流

//Reader内部类用于文件的读取操作
MapFile.Reader reader=new MapFile.Reader(fs,mapFile.toString(),conf);

//通过reader从文档中读取记录
Text key=new Text();
Text value=new Text();
while(reader.next(key,value)){
System.out.println(key);
System.out.println(key);
}
IOUtils.closeStream(reader);//关闭read流