正文
Hadoop2-HDFS学习笔记之入门（不含YARN及MR的调度功能）
小程序：扫一扫查出行【扫一扫了解最新限行尾号】
复制小程序
架构
Hadoop整体由HDFS、YARN、MapReduce三大部分组成，推荐架构参考：https://www.cnblogs.com/zhjh256/p/10573684.html。
注：2.x的时候引入了YARN、并调整了一系列进程，其性能较差，本文主要讲解2.0体系。1.0可以参考https://www.cnblogs.com/kubixuesheng/p/5525306.html。
官方文档（最好的参考资料）：http://hadoop.apache.org/docs/stable/hadoop-project-dist/hadoop-hdfs/HdfsDesign.html
整个HDFS集群由Namenode和Datanode构成master-worker（主从）模式。Namenode负责构建命名空间，管理文件的元数据等，而Datanode负责实际存储数据，负责读写工作。
名称节点，是HDFS的守护程序（一个核心程序），对整个分布式文件系统进行总控制，会纪录所有的元数据分布存储的状态信息，比如文件是如何分割成数据块的，以及这些数据块被存储到哪些节点上，还有对内存和I/O进行集中管理，用户首先会访问Namenode，通过该总控节点获取文件分布的状态信息，找到文件分布到了哪些数据节点，然后在和这些节点打交道，把文件拿到。故这是一个核心节点，发生故障将导致集群崩溃。
所以有备用名称节点、或称为 辅助名称节点，或者检查点节点 ，它是 监控HDFS状态的辅助后台程序 ，可以保存名称节点的副本，故每个集群都有一个，它与NameNode进行通讯，定期保存HDFS元数据快照。NameNode故障可以作为备用NameNode使用，CDH版本已经支持自动切换。
hdfs命令和hadoop命令的区别
可以参考https://www.cnblogs.com/lzfhope/p/6952869.html，给了较好的解释，简而言之，hdfs的相当一部分的功能可以使用hdoop来替代（目前），但hdfs有自己的一些独有的功能。hadoop主要面向更广泛复杂的功能。如果某些特性在hadoop或hdfs命令已经过时，则会给出提示，如下：
[root@quickstart ~]# hadoop fsck /user/root/hue.json -files -locations -racks #查看一个文件的详细信息，注意：此命令只能在namenode里输入，在datanode里输入会报错
DEPRECATED: Use of this script to execute hdfs command is deprecated.
Instead use the hdfs command for it. Connecting to namenode via http://quickstart.cloudera:50070/fsck?ugi=root&files=1&locations=1&racks=1&path=%2Fuser%2Froot%2Fhue.json
FSCK started by root (auth:SIMPLE) from /127.0.0.1 for path /user/root/hue.json at Sun Apr :: PDT
/user/root/hue.json bytes, block(s): OK
Status: HEALTHY
Total size: B
Total dirs:
Total files:
Total symlinks:
Total blocks (validated): (avg. block size B)
Minimally replicated blocks: (100.0 %)
Over-replicated blocks: (0.0 %)
Under-replicated blocks: (0.0 %)
Mis-replicated blocks: (0.0 %)
Default replication factor:
Average block replication: 1.0
Corrupt blocks:
Missing replicas: (0.0 %)
Number of data-nodes:
Number of racks:
FSCK ended at Sun Apr :: PDT in milliseconds The filesystem under path '/user/root/hue.json' is HEALTHY
dfs当前目录
当前目录为/user/$USER/，如下：
[root@quickstart ~]# hadoop fs -ls /user/root
Found items
-rw-r--r-- root supergroup -- : /user/root/hue.json
查看所有用户的默认目录：
[root@quickstart ~]# hadoop fs -ls /user
Found items
drwxr-xr-x - cloudera cloudera -- : /user/cloudera
drwxr-xr-x - mapred hadoop -- : /user/history
drwxrwxrwx - hive supergroup -- : /user/hive
drwxrwxrwx - hue supergroup -- : /user/hue
drwxr-xr-x - hdfs supergroup -- : /user/impala
drwxrwxrwx - jenkins supergroup -- : /user/jenkins
drwxrwxrwx - oozie supergroup -- : /user/oozie
drwxrwxrwx - root supergroup -- : /user/root
drwxr-xr-x - hdfs supergroup -- : /user/spark
写文件
在实际中，对数据文件的操作可以认为基本上都是通过java接口写到特定的目录来完成的（无论是文本文件还是Parquet文件，然后映射为Impala或Hive表）。
写文件的流程如下：
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分布在两个机架上
- 写带宽：写入管道的过程只需要跨越一个交换机
- 读带宽：可以从两个机架中任选一个读取
Java写Parquet文件示例
读写文本文件的wordcount就太没意思了，来个读写parquet的示例，直接从eclipse远程运行、而不是生成jar，放到服务器上通过hadoop jar命令运行。
package hadoop; import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.Reducer;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.input.TextInputFormat;
import org.apache.parquet.example.data.Group;
import org.apache.parquet.example.data.simple.SimpleGroupFactory;
import org.apache.parquet.hadoop.ParquetOutputFormat;
import org.apache.parquet.hadoop.example.GroupWriteSupport;
import java.io.IOException;
import java.util.Random;
import java.util.StringTokenizer;
import java.util.UUID;
/**
* * <p>Title: ParquetNewMR</p> * <p>Description: </p> * @author zjhua * @date 2019年4月7日
*/
public class ParquetNewMR { public static class WordCountMap extends
Mapper<LongWritable, Text, Text, IntWritable> { private final IntWritable one = new IntWritable(1);
private Text word = new Text();
@Override
public void map(LongWritable key, Text value, Context context)
throws IOException, InterruptedException {
String line = value.toString();
StringTokenizer token = new StringTokenizer(line);
while (token.hasMoreTokens()) {
word.set(token.nextToken());
context.write(word, one);
}
}
} public static class WordCountReduce extends
Reducer<Text, IntWritable, Void, Group> {
private SimpleGroupFactory factory;
@Override
public void reduce(Text key, Iterable<IntWritable> values,
Context context) throws IOException, InterruptedException {
int sum = 0;
for (IntWritable val : values) {
sum += val.get();
}
Group group = factory.newGroup()
.append("name", key.toString())
.append("age", sum);
context.write(null,group);
} @Override
protected void setup(Context context) throws IOException, InterruptedException {
super.setup(context);
factory = new SimpleGroupFactory(GroupWriteSupport.getSchema(context.getConfiguration())); }
} public static void main(String[] args) throws Exception {
Configuration conf = new Configuration();
String writeSchema = "message example {\n" +
"required binary name;\n" +
"required int32 age;\n" +
"}";
conf.set("parquet.example.schema",writeSchema); Job job = new Job(conf);
job.setJarByClass(ParquetNewMR.class);
job.setJobName("parquet"); String in = "hdfs://192.168.223.141:8020/user/cloudera/wordcount/input";
String out = "hdfs://192.168.223.141:8020/user/cloudera/pq_out_" + UUID.randomUUID().toString(); job.setMapOutputKeyClass(Text.class);
job.setMapOutputValueClass(IntWritable.class); job.setOutputValueClass(Group.class); job.setMapperClass(WordCountMap.class);
job.setReducerClass(WordCountReduce.class); job.setInputFormatClass(TextInputFormat.class);
job.setOutputFormatClass(ParquetOutputFormat.class); FileInputFormat.addInputPath(job, new Path(in));
ParquetOutputFormat.setOutputPath(job, new Path(out));
ParquetOutputFormat.setWriteSupportClass(job, GroupWriteSupport.class); job.waitForCompletion(true);
}
}
查看生成的文件：
读文件
读文件的流程如下：
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的内存中，因此相应位置请求非常高效，不会成为瓶颈。
Java读取Parquet文件示例
package hadoop; import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.Reducer;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
import org.apache.hadoop.mapreduce.lib.output.TextOutputFormat;
import org.apache.parquet.example.data.Group;
import org.apache.parquet.hadoop.ParquetInputFormat;
import org.apache.parquet.hadoop.api.DelegatingReadSupport;
import org.apache.parquet.hadoop.api.InitContext;
import org.apache.parquet.hadoop.api.ReadSupport;
import org.apache.parquet.hadoop.example.GroupReadSupport; import java.io.IOException;
import java.util.*; public class ParquetNewMRReader { public static class WordCountMap1 extends
Mapper<Void, Group, LongWritable, Text> { protected void map(Void key, Group value,
Mapper<Void, Group, LongWritable, Text>.Context context)
throws IOException, InterruptedException { String name = value.getString("name",0);
int age = value.getInteger("age",0); context.write(new LongWritable(age),
new Text(name));
}
} public static class WordCountReduce1 extends
Reducer<LongWritable, Text, LongWritable, Text> { public void reduce(LongWritable key, Iterable<Text> values,
Context context) throws IOException, InterruptedException {
Iterator<Text> iterator = values.iterator();
while(iterator.hasNext()){
context.write(key,iterator.next());
}
} } public static final class MyReadSupport extends DelegatingReadSupport<Group> {
public MyReadSupport() {
super(new GroupReadSupport());
} @Override
public org.apache.parquet.hadoop.api.ReadSupport.ReadContext init(InitContext context) {
return super.init(context);
}
} public static void main(String[] args) throws Exception {
Configuration conf = new Configuration();
String readSchema = "message example {\n" +
"required binary name;\n" +
"required int32 age;\n" +
"}";
conf.set(ReadSupport.PARQUET_READ_SCHEMA, readSchema); Job job = new Job(conf);
job.setJarByClass(ParquetNewMRReader.class);
job.setJobName("parquet"); String in = "hdfs://192.168.223.141:8020/user/cloudera/pq_out_ae7d1402-4c53-45b7-bf10-54e05fcdeb58";
String out = "hdfs://localhost:8020/user/cloudera/wd2"; job.setMapperClass(WordCountMap1.class);
job.setReducerClass(WordCountReduce1.class); job.setInputFormatClass(ParquetInputFormat.class);
ParquetInputFormat.setReadSupportClass(job, MyReadSupport.class);
ParquetInputFormat.addInputPath(job, new Path(in)); job.setOutputFormatClass(TextOutputFormat.class);
FileOutputFormat.setOutputPath(job, new Path(out)); job.waitForCompletion(true);
}
}
生成的文件如下：
创建hive表进行验证。。。。
INFO : Completed executing command(queryId=hive_20190420132929_1fc94be6-c915-4897-babf-4d0cf52ecdbb); Time taken: 0.029 seconds
INFO : OK
+-------------------+--+
| tab_name |
+-------------------+--+
| parquet_xxx |
| test_parquet_new |
+-------------------+--+
2 rows selected (0.206 seconds)
0: jdbc:hive2://localhost:10000/default> select * from test_parquet_new;
INFO : Compiling command(queryId=hive_20190420132929_9097c88f-f390-4fa2-891c-569b03fe106d): select * from test_parquet_new
INFO : Semantic Analysis Completed
INFO : Returning Hive schema: Schema(fieldSchemas:[FieldSchema(name:test_parquet_new.name, type:string, comment:null), FieldSchema(name:test_parquet_new.age, type:int, comment:null)], properties:null)
INFO : Completed compiling command(queryId=hive_20190420132929_9097c88f-f390-4fa2-891c-569b03fe106d); Time taken: 0.11 seconds
INFO : Executing command(queryId=hive_20190420132929_9097c88f-f390-4fa2-891c-569b03fe106d): select * from test_parquet_new
INFO : Completed executing command(queryId=hive_20190420132929_9097c88f-f390-4fa2-891c-569b03fe106d); Time taken: 0.001 seconds
INFO : OK
+------------------------+-----------------------+--+
| test_parquet_new.name | test_parquet_new.age |
+------------------------+-----------------------+--+
| Hadoop | 3 |
| Oh | 1 |
| a | 1 |
| an | 1 |
| as | 2 |
| be | 1 |
| can | 1 |
| elephant | 1 |
| fellow | 1 |
| is | 3 |
| what | 1 |
| yellow | 2 |
+------------------------+-----------------------+--+
12 rows selected (0.788 seconds)
读写部分原理引用了：Hadoop权威指南第3章。
parquet使用：https://github.com/apache/parquet-mr/
