HDFS文件读写操作

布满荆棘的人生 2022-05-28 01:58 220阅读 0赞

**HDFS读文件**

1.  客户端首先调用FileSystem对象的open方法打开文件，其实获取的是一个DistributedFileSystem的实例。
2.  DistributedFileSystem通过调用RPC(远程过程调用)向namenode发起请求，获得文件的第一批block的位置信息。同一block按照备份数会返回多个DataNode的位置信息，并根据集群的网络拓扑结构排序，距离客户端近的排在前面， 如果客户端本身就是该DataNode，那么它将从本地读取文件。
3.  DistributedFileSystem类返回一个FSDataInputStream对象给客户端，用来读取数据，该对象会被封装成DFSInputStream对象，该DFSInputStream对象管理着datanode和namenode的I/O数据流。客户端对输入端调用read方法，DFSInputStream就会找出离客户端最近的datanode并连接datanode。
4.  在数据流中重复调用read()函数，直到这个块全部读完为止。DFSInputStream关闭和此DataNode的连接。接着读取下一个bloc块。这些操作对客户端来说是透明的，从客户端的角度来看只是读一个持续不断的流。  
    每读取完一个block都会进行checksum验证，如果读取datanode时出现错误，客户端会通知Namenode，然后再从下一个拥有该block拷贝的datanode继续读。
5.  当正确读取完当前block的数据后，关闭当前的DataNode链接，并为读取下一个block寻找最佳的DataNode。如果第一批block都读完了，且文件读取还没有结束，DFSInputStream就会去namenode拿下一批block的位置信息继续读。
6.  当客户端读取完毕数据的时候，调用FSDataInputStream的close方法关闭掉所有的流。

【注意】  
在读取数据的过程中，如果客户端在与数据节点通信出现错误，则尝试连接包含此数据块的下一个数据节点。同时会记录这个节点的故障。这样它就不会再去尝试连接和读取块。客户端还会验证从DataNode传送过来的数据校验和。如果发现一个损坏的块，那么客户端将会再尝试从别的DataNode读取数据块，向NameNode报告这个信息，NameNode也会更新保存的文件信息。

这里要关注的一个设计要点是，客户端通过NameNode引导获取最合适的DataNode地址，然后直接连接DataNode读取数据。这种设计的好处是，可以使HDFS扩展到更大规模的客户端并行处理，这是因为数据的流动是在所有DataNode之间分散进行的。同时NameNode的压力也变小了，使得NameNode只用提供请求块所在的位置信息就可以了，而不用通过它提供数据，这样就避免了NameNode随着客户端数量的增长而成为系统瓶颈。

![这里写图片描述][70]

**使用FileSystem读取文件**

调用open()函数来获取文件的输入流，有以下两种方法
        Public FSDataInputStream open(Path f) throws IOException
        Public abstract FSDataInputStream open(Path f, int bufferSize) throws IOException
    
    public class FileSystemCat {
        public static void main(String[] args) throws Exception {
            String uri = args[0];
            Configuration conf = new Configuration();   //拿到HDFS文件系统中的URI
            FileSystem fs = FileSystem.get(URI.create(uri), conf);
            InputStream in = null;
            try {
                in = fs.open(new Path(uri));        //返回一个FSDatalnputStream
                IOUtils.copyBytes(in, System.out, 4096, false);
            } finally {
                IOUtils.closeStream(in);
            }
        }
    }
    
    IOUtils.copyBytes()
        in表示拷贝源
        System.out表示拷贝目的地（也就是要拷贝到标准输出中去）
        4096表示用来拷贝的buffer大小
        false表明拷贝完后不关闭拷贝源和拷贝目的地（因为System.out不需要关闭，in可以在finally语句中关闭）
    
    IOUtils.closeStream()，用来关闭一个流。

**HDFS写文件**

1.  客户端通过调用DistributedFileSystem的create方法，创建一个新的文件
2.  DistributedFileSystem通过RPC（远程过程调用）向NameNode发起请求，去创建一个没有block关联的新文件，创建前，NameNode会做各种校验，比如文件是否存在，客户端有无权限去创建等。如果校验通过，NameNode就会进行记录，并返回文件的block列表（所有的副本）对应的DataNode地址信息，否则就会抛出IO异常
3.  DistributedFileSystem返回FSDataOutputStream的对象，用于客户端写数据，FSDataOutputStream被封装成DFSOutputStream，DFSOutputStream可以协调NameNode和DataNode。客户端开始写数据到DFSOutputStream，DFSOutputStream会把数据切成一个个packet包，以数据队列“data queue”的形式管理这些packet
4.  DataStreamer接受并处理data queue，向Namenode申请blocks，获取用来存储replicas的合适的datanode列表，把它们排成一个pipeline管道（列表的大小根据Namenode中replication的设定而定）。DataStreamer把packet按顺序输出到管道的第一个DataNode中，将该packet存储之后，再将其传递给在此pipeline中的下一个datanode，直到最后一个datanode，这种写数据的方式呈流水线的形式。
    
    只要写入了dfs．replication．min的复本数（默认为1），写操作就会成功，并且这个块可以在集群中异步复制，直到达到其目标复本数（dfs．replication的默认值为3），因为namenode已经知道文件由哪些块组成，所以它在返回成功前只需要等待数据块进行最小量的复制。
5.  在最后一个datanode成功存储之后会返回一个ack packet（确认队列），也是由packet组成（保存发出去的数据块）。DataStreamer将所有的数据块都刷到pipeline中的数据节点， 然后等待ack queue返回成功，当客户端成功收到datanode返回的ack packet后，通知 DataNode 把文件标示为已完成，并从”ack queue”移除相应的packet。
    
    如果传输过程中，有某个datanode出现了故障：  
    那么当前的pipeline会被关闭，出现故障的datanode会从当前的pipeline中移除。将ack queue中的数据块放入data queue的开始，文件会继续被写到pipeline管道中剩余的DataNode中，因为此数据块的副本数没有达到配置要求，Namenode会分配一个新的datanode，随后的文件会正常执行写入操作，保持replicas设定的数量。
6.  当客户端结束写入数据，调用stream的close()方法关闭数据流

![这里写图片描述][70 1]

**使用FileSystem写入数据**

指定一个Path对象然后返回一个用于写入数据的输出流

public FSDataOutputStream create(Path f)throws IOException
        create()方法能够为需要写入且当前不存在的文件创建父目录
    public FSDataOutputStream append(Path f)throws IOException
        append()方法在一个已有文件末尾追加数据
    
    将本地文件复制到Hadoop文件系统-create
        public class FileCopy{
            public static void main(String[] args) throws Exception {
                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);
                FSDataOutputStream out = fs.create(new Path(dst));  
                IOUtils.copyBytes(in, out, 4096, true);
            }
        }   
    
    将本地文件复制到Hadoop文件系统-append
    public class CopyFileAppend {
        public static void main(String[] args) {
            String localSrc = args[0];
            String dist = args[1];
            BufferedInputStream in = null;
            try{
                in = new BufferedInputStream(new FileInputStream(localSrc));
                Configuration conf = new Configuration();
                FileSystem fs = FileSystem.get(URI.create(dist), conf);
                FSDataOutputStream out = fs.append(new Path(dist));
                IOUtils.copyBytes(in, out, 4096, false);
            }catch(Exception e){
                e.printStackTrace();
            }finally{
                IOUtils.closeStream(in);
            }
        }
    }

[70]: /images/20220528/9951feed8a1b4ac19adc12eaf1458699.png
[70 1]: /images/20220528/a6620965d8494008b9c28d93f9242f65.png