hadoop 学习笔记（2）-- hadoop 文件系统

2019年6月8日 218次阅读

当数据量很大，一台物理机无法容纳时，我们就需要将数据存储到由网络连接的若干台机器上，这就是所谓的分布式文件系统。hadoop 使用 hdfs 作为自己的分布式文件系统。目前 hdfs 不适合低延迟访问，也不适合存储大量的小文件以及随机写入（hdfs 中只能在文件末尾进行写操作）。

1 hdfs 的概念

硬盘有自己的块大小，表示其可以读写的最小数据量。文件系统块的大小为若干倍的硬盘大小。
hdfs 上的文件也以块来划分，默认为 128M，为了容错，每个 block 通常在其他节点有若干个备份，一个节点出问题了可以使用别的节点的数据。

使用下面的命令可以查看 hdfs 的信息：

hdfs fsck / -files -blocks

2 namenode 与 datanode

hdfs 集群包括两种节点：

  • namenode：是集群的管理者，记录着文件系统树以及所有文件和目录的元数据。namenode 还记录着组成文件的 block 的位置（在每次系统启动时建立）
  • datanode：存储数据并提供数据，定时向 namenode 同步自己存储的 block 的信息

经常被读取的文件的相关 block 会被缓存在 datanode 的内存中，用户也可以指定哪些文件的 block 应当被缓存。

3 命令行接口

在学习 hadoop 时，我们使用伪分布式的方法启动 hadoop 集群。在 **core-site.xml ** 中可以通过设置下面的属性来配置默认的文件系统。

<property>
       <name>fs.defaultFS</name>
       <value>hdfs://localhost:9000</value>
</property>

基础的操作有：

//将本地文件系统的数据复制到 hdfs 中，由于已经配置了 fs.defaultFS 可以省略 hdfs://localhost:9000
hdfs dfs -copyFromLocal /test/data/* /test1/input
//使用相对路径会将数据复制到用户目录下 我这里是 /user/hadoop
hdfs dfs -copyFromLocal /test/data/* input
//从hdfs copy 到本地
hdfs dfs -copyToLocal test.txt text.txt
//创建文件夹
hdfs dfs -mkdir books
//列出所有文件
hdfs dfs -ls .
//print 文件的内容
hdfs dfs -cat /user/hadoop/books

hdfs 的文件也有 r w x 的概念，当然这里的 x 是指是否可以访问一个目录下面的文件。hdfs 中的文件也有所属 user 以及 group 的概念，使用当前调用者进程的 user 和 group 来标识，运行 namenode 的用户有 “root” 权限。

hadoop 有一个抽象的文件系统，hdfs 是其实现之一，通过 uri 的方式，可以选择不同的文件系统，比如hdfs

hdfs dfs -ls file:///home/hadoop

4 java api

使用 URL 就可以访问 hdfs 文件，不过要进行一点小处理，设置 UrlStreamHandlerFactory：

《hadoop 学习笔记（2）-- hadoop 文件系统》

不方便设置 UrlStreamHandlerFactory 时，可以使用 FileSystem Api：

public class ReadFile {
    public static void main(String[] args) throws IOException {
        if (args.length != 1) {
            System.out.println("Usage: ReadFile path/to/file");
            System.exit(-1);
        }
        String filePath = args[0];
        if (!filePath.startsWith("/")) {
            filePath = "/" + filePath;
        }
        filePath = "hdfs://localhost:9000" + filePath;
        // conf 会根据 classpath 中的 core-site.xml 创建
        Configuration conf = new Configuration();
        //创建 fileSystem 实例
        FileSystem fileSystem = FileSystem.get(URI.create(filePath), conf);
        InputStream is = null;
        try {
            //使用 open 打开 InputStream
            is = fileSystem.open(new Path(filePath));
            IOUtils.copyBytes(is, System.out, 4096, false);
        } finally {
            IOUtils.closeStream(is);
        }
    }
}

其中，FileSystem.open 返回的是一个 FSDataInputStream 对象，可以使用 seek 调整读取的位置：

《hadoop 学习笔记（2）-- hadoop 文件系统》

另外，其还实现了 PositionedReadable 接口，可以从指定位置读取指定的数据：

public interface PositionedReadable {
  /**
   * Read upto the specified number of bytes, from a given
   * position within a file, and return the number of bytes read. This does not
   * change the current offset of a file, and is thread-safe.
   */
  public int read(long position, byte[] buffer, int offset, int length)
    throws IOException;
  
  /**
   * Read the specified number of bytes, from a given
   * position within a file. This does not
   * change the current offset of a file, and is thread-safe.
   */
  public void readFully(long position, byte[] buffer, int offset, int length)
    throws IOException;
  
  /**
   * Read number of bytes equal to the length of the buffer, from a given
   * position within a file. This does not
   * change the current offset of a file, and is thread-safe.
   */
  public void readFully(long position, byte[] buffer) throws IOException;
}

使用 FileSystem api 也可以进行写操作：

        String localFile = args[0];
        String hdfsFile = args[1];
        if (!hdfsFile.startsWith("/")) {
            hdfsFile = "/" + hdfsFile;
        }
        hdfsFile = "hdfs://localhost:9000" + hdfsFile;

        InputStream in = null;
        FSDataOutputStream out = null;
        try {
            in = new BufferedInputStream(new FileInputStream(localFile));

            FileSystem fileSystem = FileSystem.get(URI.create(hdfsFile), new Configuration());
            out = fileSystem.create(new Path(hdfsFile));

            IOUtils.copyBytes(in, out, 4096, true);
        } catch (Exception e) {
            e.printStackTrace();
        }

通过 FileSystem.getStatus 可以查看文件的状态。

通过 FileSystem.listStatus 可以获取文件夹下的所有文件的信息。

通过 FileSystem.globStatus 可以按照一定的 pattern 查看

通过 PathFilter 可以在globStatus 加入更细粒度的控制:

fs.globStatus(new Path("/2007/*/*"), new RegexExcludeFilter("^.*/2017/12/31$"));

删除数据使用 FileSystem 的 delete 方法进行。

文件读取的过程中，从 namenode 获取 block 的位置，再去对应的 datanode 读取数据：

《hadoop 学习笔记（2）-- hadoop 文件系统》 文件读取

文件写入数据的过程中，先在 namenode 创建元数据，然后向一个 datanode 写入数据，该 datanode 会同时向其他 datanode 同步数据。

《hadoop 学习笔记（2）-- hadoop 文件系统》 写入数据

数据写入时，应该在需要的地方调用输出流的 sync，以便于将缓存写入文件系统中，注意 close 时也会调用一次 sync。

5 distcp

使用 distcp 可以并发的进行数据复制，个人认为功能可以与 rsync 类比。

hadoop@millions-server:/usr/local/hadoop/etc/hadoop$ hdfs dfs -ls testdata
Found 1 items
-rw-r–r– 1 hadoop supergroup 288374 2017-03-07 23:22 testdata/synthetic_control.data

而后输入：

hadoop distcp -update testdata/synthetic_control.data testdata/synthetic_control2.data

可以看到：

hadoop@millions-server:/usr/local/hadoop/etc/hadoop$ hdfs dfs -ls testdata
Found 2 items
-rw-r–r– 1 hadoop supergroup 288374 2017-03-07 23:22 testdata/synthetic_control.data
-rw-r–r– 1 hadoop supergroup 288374 2017-05-06 23:09 testdata/synthetic_control2.data

再来同步一个文件夹：

hadoop distcp testdata testdata2
hdfs dfs -ls testdata2
Found 2 items
-rw-r–r– 1 hadoop supergroup 288374 2017-05-06 23:14 testdata2/synthetic_control.data
-rw-r–r– 1 hadoop supergroup 288374 2017-05-06 23:14 testdata2/synthetic_control2.data

删除 testdata 中的 synthetic_control2.data：

hdfs dfs -rm testdata/synthetic_control2.data

再进行一次带 -update 和 -delete 的 distcp，可以看到，多余的文件被删除了：

hadoop distcp -update -delete testdata testdata2
hdfs dfs -ls testdata2
Found 1 items
-rw-r–r– 1 hadoop supergroup 288374 2017-05-06 23:14 testdata2/synthetic_control.data

点赞