安装Hadoop

Docker安装Hadoop

Linux服务器搭建Hadoop3.X完全分布式集群环境

HDFS架构设计

HDFS采用master/slave架构。一个HDFS集群是由一个Namenode和一定数目的Datanodes组成。

Namenode是一个中心服务器，负责管理文件系统的名字空间(namespace)以及客户端对文件的访问。

集群中的Datanode一般是一个节点一个，负责管理它所在节点上的存储。HDFS暴露了文件系统的名字空间，用户能够以文件的形式在上面存储数据。从内部看，一个文件其实被分成一个或多个数据块，这些块存储在一组Datanode上。

Namenode执行文件系统的名字空间操作，比如打开、关闭、重命名文件或目录。它也负责确定数据块到具体Datanode节点的映射。Datanode负责处理文件系统客户端的读写请求。在Namenode的统一调度下进行数据块的创建、删除和复制。

API操作

HDFS的开发主要是在客户端，其核心是从HDFS提供的API中构造一个HDFS的访问客户端对象，然后通过该客户端对象操作HDFS上的文件

添加依赖

 <dependencies>
        <dependency>
            <groupId>org.apache.hadoop</groupId>
            <artifactId>hadoop-common</artifactId>
            <version>3.3.1</version>
        </dependency>
        <dependency>
            <groupId>org.apache.hadoop</groupId>
            <artifactId>hadoop-hdfs</artifactId>
            <version>3.3.1</version>
        </dependency>
        <dependency>
            <groupId>org.apache.hadoop</groupId>
            <artifactId>hadoop-hdfs-client</artifactId>
            <version>3.3.1</version>
        </dependency>
        <dependency>
            <groupId>org.apache.hadoop</groupId>
            <artifactId>hadoop-client</artifactId>
            <version>3.3.1</version>
        </dependency>
        <dependency>
            <groupId>junit</groupId>
            <artifactId>junit</artifactId>
            <version>4.13</version>
            <scope>test</scope>
        </dependency>
    </dependencies>

获取FileSystem

FileSystem对象是一个文件系统对象, 可以用该对象的一些方法来对文件进行操作, 通过FileSystem的静态方法get或newInstance获得该对象。

4种方式获取FileSystem

    @Test
    public void getFileSystem1() throws URISyntaxException, IOException {
    	// 封转客户端或者服务器的配置
        Configuration configuration = new Configuration();
        // 设置要操作的文件系统
        configuration.set("fs.defaultFS","hdfs://IP:9000");
        // 获取指定的文件系统
        FileSystem fileSystem = FileSystem.get(new URI("/"), configuration);
//        FileSystem fileSystem = FileSystem.newInstance(configuration);
        System.out.println(fileSystem.toString());
    }

    @Test
    public void getFileSystem2() throws URISyntaxException, IOException {
        Configuration configuration = new Configuration();
        FileSystem fileSystem = FileSystem.get(new URI("hdfs://IP:9000"), configuration);
//        FileSystem fileSystem = FileSystem.newInstance(new URI("hdfs://IP:9000"), configuration);
        System.out.println(fileSystem.toString());
    }

遍历所有文件

	@Test
    public void listFile2() throws Exception {
        FileSystem fileSystem = FileSystem.get(new URI("hdfs://IP:9000"), new Configuration());
        //获取所有的文件或者文件夹; 指定遍历的路径，指定是否要递归遍历
        RemoteIterator<LocatedFileStatus> locatedFileStatusRemoteIterator = fileSystem.listFiles(new Path("/"), true);
        while (locatedFileStatusRemoteIterator.hasNext()) {
            // 获取得到每一个文件详细信息
            LocatedFileStatus fileStatus = locatedFileStatusRemoteIterator.next();
            // 获取每一个文件存储路径 名称
            System.out.println(fileStatus.getPath());
            System.out.println(fileStatus.getPath().getName());

            // 所有文件的Block存储信息
            BlockLocation[] blockLocations = fileStatus.getBlockLocations();
            // 每个文件的Block数量
            System.out.println("blockLocations = " + blockLocations.length);
            // 每一个Block副本存储位置
            for (BlockLocation blockLocation : blockLocations) {
                String[] hosts = blockLocation.getHosts();
                for (String host : hosts) {
                    System.out.println("host = " + host);
                }
            }
        }
        fileSystem.close();
    }

文件权限问题

对HDFS文件读写时可能遇到权限问题，解决方案如下

1.HDFS的权限配置，关闭或开启后重启

cd	/usr/local/hadoop/etc/hadoop

vi hdfs-site.xml

<property>
    <name>dfs.permissions.enabled</name>
    <value>true</value>
</property>

2.使用HDFS的命令修改相应目录的权限

hadoop fs -chmod 777 /user /user是上传文件的路径

3.获取fileSystem 时指定以某用户的身份去访问

FileSystem fileSystem = FileSystem.get(new URI("hdfs://IP:9000"), new Configuration(),"root");

创建文件夹及文件

    @Test
    public void mkdirsOrCreate() throws  Exception{
        FileSystem fileSystem = FileSystem.get(new URI("hdfs://IP:9000"), new Configuration());
        fileSystem.mkdirs(new Path("/user/test"));
        fileSystem.create(new Path("/user/test/test.txt"));
        fileSystem.close();
    }

删除文件夹及文件

    @Test
    public  void  deleteFile() throws Exception{
        FileSystem fileSystem = FileSystem.get(new URI("hdfs://IP:9000"), new Configuration(), "root");
        fileSystem.delete(new Path("/user/test/test.txt"),false);
        fileSystem.delete(new Path("/user/test"),true);
        fileSystem.close();
    }

文件上传

@Test
    public void uploadFile() throws Exception {
        FileSystem fileSystem = FileSystem.get(new URI("hdfs://IP:9000"), new Configuration());
        fileSystem.copyFromLocalFile(new Path("file:///G:\\test.txt"), new Path("/user/test/test2.txt"));
        fileSystem.close();
    }

    @Test
    public void uploadFile2() throws Exception {
        FileSystem fileSystem = FileSystem.get(new URI("hdfs://IP:9000"), new Configuration());
        FSDataOutputStream fsDataOutputStream = fileSystem.create(new Path("/user/test/test3.txt"));
        FileInputStream fileInputStream = new FileInputStream(new File("G:\\test.txt"));
        // 实现文件的复制
        IOUtils.copy(fileInputStream, fsDataOutputStream);
        // 关闭流
        IOUtils.closeQuietly(fileInputStream);
        IOUtils.closeQuietly(fsDataOutputStream);
        fileSystem.close();
    }

文件下载

    @Test
    public void downloadFile()throws  Exception{
        FileSystem fileSystem = FileSystem.get(new URI("hdfs://IP:9000"), new Configuration());
        // 获取HDFS文件输入流
        FSDataInputStream open = fileSystem.open(new Path("/user/test/test.txt"));
        // 获取本地文件输出流
        FileOutputStream fileOutputStream = new FileOutputStream(new File("G:\\test.txt"));
        // 实现文件的复制
        IOUtils.copy(open,fileOutputStream );
        // 关闭流
        IOUtils.closeQuietly(open);
        IOUtils.closeQuietly(fileOutputStream);
        fileSystem.close();
    }

   @Test
    public void downloadFile2()throws  Exception{
        FileSystem fileSystem = FileSystem.get(new URI("hdfs://IP:9000"), new Configuration());
        fileSystem.copyToLocalFile(new Path("/user/test/test.txt"),new Path("G://test.txt"));
        fileSystem.close();
    }

小文件合并

在Hadoop中，每个文件都需要维护一份元数据信息，而Hadoop擅长存储大文件，因为大文件的元数据信息比较少。如果Hadoop集群当中存在大量小文件，会大大增加集群管理元数据的内存压力，所以有必要将小文件合并成大文件进行一起处理。

HDFS的Shell命令模式下，将hdfs文件合并成一个大文件进行下载

hdfs dfs -getmerge /user/test/*.txt ./heBing.txt

在上传的时候将小文件合并到一个大文件

 @Test
    public void mergeFile() throws Exception {
        //获取分布式文件系统
        FileSystem fileSystem = FileSystem.get(new URI("hdfs://IP:9000"), new Configuration(), "root");
        FSDataOutputStream outputStream = fileSystem.create(new Path("/user/test/bigfile.txt"));
        //获取本地文件系统
        LocalFileSystem local = FileSystem.getLocal(new Configuration());
        //通过本地文件系统获取文件列表，为一个集合
        FileStatus[] fileStatuses = local.listStatus(new Path("file:///G:\\test"));
        for (FileStatus fileStatus : fileStatuses) {
            FSDataInputStream inputStream = local.open(fileStatus.getPath());
            IOUtils.copy(inputStream, outputStream);
            IOUtils.closeQuietly(inputStream);
        }
        IOUtils.closeQuietly(outputStream);
        local.close();
        fileSystem.close();
    }