hdfs写数据流程
- 将文件添加到目录树
- 将数据写入队列
- 机架感知
- DN写数据、客户端收到应答
将文件添加到目录树
假设我们有一段创建文件并写入数据的代码：
FileSystem fileSystem;
@Before
public void init() throws Exception{
String user = "ocean";
Configuration configuration = new Configuration();
fileSystem = FileSystem.get(new URI("hdfs://hadoop102:8020"), configuration, user);
}
@After
public void close() throws Exception{
if(fileSystem != null){
fileSystem.close();
}
}
/**
* 测试创建文件并写入数据
* @throws IOException
*/
@Test
public void testCreatePathAndWrite() throws IOException {
FSDataOutputStream fsDataOutputStream = fileSystem.create(new Path("/path"));
fsDataOutputStream.write("hello".getBytes());
}
我们点进
fileSystem.create(new Path("/path"));
一路create，直到
再到流的创建：
最后我们回到NN的RPCServer，客户端要向NameNode的RPCServer发送请求了：
进startFile：
一直进startFile，直到找到addFile：
addINode就是添加目录树。
在此就会添加文件。
再次回到客户端请求创建目录树那里：
他现在要往外写数据。
这个stream是一个线程。
现在就有一个队列在这里阻塞。
将数据写入队列
现在我们看如何往外写数据。
找到FSOutputSummer的write。
往外写chunk和chunkSum。
将数据和校验和写到Packet中。
写了之后再看检查队列是否已满。
如果队列满了，就要等。
现在我们的队列肯定是没满。
所以走queuePacket的操作。
现在是将Packet中的数据放到队列中，并且唤醒队列。
还记得我们第一部分有一个队列在等待吗？现在他醒了。
机架感知
第一部分的阻塞走通了。
现在他要建立管道，其实就是要知道往那几个DN写。
我们进入nextBlockOutputStream()。
到底向哪个DN写。
到了NN的RPCServer了。
然后一路chooseTarget知道最后的策略。
进入按顺序获取目的地。
第一个副本选择当前节点，第二个副本选择与当前机架不同的节点。
第三个副本放在和dn1一样的机架。
DN写数据、客户端收到应答
再次回到获取DN那里：
获取之后，我们就要往第一个DN写，进入createBlockOutputStream。
这里他建立了输出流和输入流，分别是用来往DN写，和从DN接受信息。
发送数据。
执行发送。
我们现在的操作是WRITE_BLOCK。
找一个类：
这个DN这边用来接受数据的类。
他在run方法中又new了一个线程，进去查看。
找到它的run方法。
解析操作并处理操作。
还记得我们的操作是WRITE_BLOCK？进去。
首先往当前DN写一份临时数据。
然后再往其他DN发送消息。比如DN0通知DN1，这时候DN1也会往磁盘写一份临时数据。
如此往复下去。
现在要往回应答。比如有三个DN，DN2告诉DN1写好了，DN1告诉DN0写好了。我们要接受这个应答。
再次回到DataStreamer的run方法。
进入initDataStreaming()。
获取应答。
ResponseProcessor是一个线程，走它的run方法。
如果数据往DN发送成功了，会将其从ackQueue中移除。但是我们不知道ackQueue是什么。我们知道有一个dataQueue，他的数据往DN走。所以这里相当于是用ackQueue来做一个监听。
重新回到initDataStreaming()。
他把dataQueue中的packet挪到了ackQueue。
为什么要往ackQueue放呢，因为怕写失败了，写失败了的话，就再添到dataQueue再次往DN写。
如果全部写成功了，ackQueue中的数据包就可以干掉了。
