HDFS

Posted 2021-04-14 虹猫tomcat

HDFS的API操作、HDFS的高可用机制以及Hadoop的联邦机制。

HDFS的API操作

环境准备

1 拷贝hadoop2.7.5至无空格无中文的路径下。

2 配置环境变量并添加至path中。

3 将hadoop2.7.5/bin下的hadoop.dll添加至C:WindowsSystem32下。

4 重启系统

5 在IDEA中新建项目导入依赖：

<?xml version="1.0" encoding="UTF-8"?><project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd"> <modelVersion>4.0.0</modelVersion>
 <groupId>com.zhu</groupId> <artifactId>hdfsstudy</artifactId> <version>1.0-SNAPSHOT</version> <dependencies> <dependency> <groupId>org.apache.hadoop</groupId> <artifactId>hadoop-common</artifactId> <version>2.7.5</version> </dependency> <dependency> <groupId>org.apache.hadoop</groupId> <artifactId>hadoop-client</artifactId> <version>2.7.5</version> </dependency> <dependency> <groupId>org.apache.hadoop</groupId> <artifactId>hadoop-hdfs</artifactId> <version>2.7.5</version> </dependency> <dependency> <groupId>org.apache.hadoop</groupId> <artifactId>hadoop-mapreduce-client-core</artifactId> <version>2.7.5</version> </dependency> <dependency> <groupId>junit</groupId> <artifactId>junit</artifactId> <version>4.12</version> </dependency> </dependencies> <build> <plugins> <plugin> <groupId>org.apache.maven.plugins</groupId> <artifactId>maven-compiler-plugin</artifactId> <version>3.1</version> <configuration> <source>1.8</source> <target>1.8</target> <encoding>UTF-8</encoding> </configuration> </plugin> <plugin> <groupId>org.apache.maven.plugins</groupId> <artifactId>maven-shade-plugin</artifactId> <version>2.4.3</version> <executions> <execution> <phase>package</phase> <goals> <goal>shade</goal> </goals> <configuration> <minimizeJar>true</minimizeJar> </configuration> </execution> </executions> </plugin> </plugins> </build></project>

访问数据

访问数据有两种方式：使用URL方式访问数据、使用文件系统方式访问数据。

使用URL方式访问数据

@Test public void demo01() throws IOException { URL.setURLStreamHandlerFactory(new FsUrlStreamHandlerFactory()); InputStream inputStream = new URL("hdfs://node01:8020/d.txt").openStream(); FileOutputStream outputStream = new FileOutputStream(new File("E:\hello.txt")); IOUtils.copyBytes(inputStream,outputStream,1024); IOUtils.closeStream(inputStream); IOUtils.closeStream(outputStream);}

运行程序，出现警告信息，不影响结果

可以在resources目录下添加log4j.properties配置文件来消除警告：

# Configure logging for testing: optionally with log file
#log4j.rootLogger=debug,appenderlog4j.rootLogger=info,appender #log4j.rootLogger=error,appender
#u8F93u51FAu5230u63A7u5236u53F0log4j.appender.appender=org.apache.log4j.ConsoleAppender #u6837u5F0Fu4E3ATTCCLayoutlog4j.appender.appender.layout=org.apache.log4j.TTCCLayout

查看结果：

使用文件系统方式访问数据

文件系统是对磁盘文件的一种管理方式，访问本地磁盘就是本地文件系统，访问HDFS就是分布式文件系统。使用文件系统访问数据有以下4种方式：

@Testpublic void getFileStream1() throws IOException { Configuration configuration = new Configuration(); //指定文件系统类型 configuration.set("fs.dafaultFS","hdfs://node01:8020/"); //获取指定的文件系统 FileSystem fileSystem = FileSystem.get(configuration); System.out.println(fileSystem.toString());}

@Testpublic void getFileStream2() throws IOException { Configuration configuration = new Configuration(); configuration.set("fs.dafaultFS","hdfs://node01:8020/"); FileSystem fileSystem = FileSystem.newInstance(configuration); System.out.println(fileSystem.toString());}

@Testpublic void getFileStream3() throws Exception { FileSystem fileSystem = FileSystem.get(new URI("hdfs://node01:8020/"),new Configuration()); System.out.println(fileSystem.toString());}

@Testpublic void getFileStream4() throws Exception { FileSystem fileSystem = FileSystem.newInstance(new URI("hdfs://node01:8020/"),new Configuration()); System.out.println(fileSystem.toString());}

遍历HDFS中文件

@Testpublic void listFiles() throws IOException, URISyntaxException { FileSystem fileSystem = FileSystem.get(new URI("hdfs://node01:8020/"),new Configuration()); RemoteIterator<LocatedFileStatus> iterator = fileSystem.listFiles(new Path("/"), true); LocatedFileStatus fileStatus = iterator.next(); //获取文件块数 BlockLocation[] blockLocations = fileStatus.getBlockLocations(); System.out.println(blockLocations.length); //获取文件名 String name = fileStatus.getPath().getName(); System.out.println(name);}

HDFS上创建文件夹、文件

@Testpublic void mk() throws IOException, URISyntaxException { FileSystem fileSystem = FileSystem.get(new URI("hdfs://node01:8020/"), new Configuration()); boolean mkdirs = fileSystem.mkdirs(new Path("/aa/bb/cc")); boolean mkdirs1 = fileSystem.mkdirs(new Path("/zhu.txt")); fileSystem.close();}

下载文件

方式一

@Testpublic void download() throws IOException, URISyntaxException { FileSystem fileSystem = FileSystem.get(new URI("hdfs://node01:8020/"), new Configuration()); InputStream inputStream = fileSystem.open(new Path("hdfs://node01:8020/d.txt")); FileOutputStream outputStream = new FileOutputStream(new File("E:\hehe.txt")); IOUtils.copyBytes(inputStream,outputStream,1024); IOUtils.closeStream(inputStream); IOUtils.closeStream(outputStream); fileSystem.close();}

方式二

@Testpublic void download() throws IOException, URISyntaxException {        FileSystem fileSystem = FileSystem.get(new URI("hdfs://node01:8020/"), new Configuration()); fileSystem.copyToLocalFile(new Path("/d.txt"),new Path("E:\hehe1.txt")); fileSystem.close();}

上传文件

@Test public void upload() throws IOException, URISyntaxException { FileSystem fileSystem = FileSystem.get(new URI("hdfs://node01:8020/"), new Configuration()); fileSystem.copyFromLocalFile(new Path("E:\application.yml"),new Path("/aaa/aaa1")); fileSystem.close(); }

HDFS访问权限控制

修改文件权限

发现依然可以下载到本地，这是因为配置文件中的dfs.permissions设置为了false，权限设置不生效。

1. 停止hdfs集群，在node01机器上执行以下命令

cd /export/servers/hadoop-2.7.5sbin/stop-dfs.sh

2. 修改node01机器上的hdfs-site.xml配置文件

cd /export/servers/hadoop-2.7.5/etc/hadoopvim hdfs-site.xml

3. 修改完成之后将配置文件发送到其他机器

scp hdfs-site.xml node02:$PWDscp hdfs-site.xml node03:$PWD

4. 重启hdfs集群

cd /export/servers/hadoop-2.7.5sbin/start-dfs.sh

再次下载，报错!

说明修改权限生效。

再次给文件修改权限为所有者有读写权限(600)，并下载文件，发现依然报错。

这时，要想访问数据，有两种方式。

第一种，权限设为全部可读可写(666)。

第二种，虚拟用户为root。

@Testpublic void download() throws IOException, URISyntaxException, InterruptedException {    FileSystem fileSystem = FileSystem.get(new URI("hdfs://node01:8020/"), new Configuration(),"root"); fileSystem.copyToLocalFile(new Path("/d.txt"),new Path("E:\hehe3.txt")); fileSystem.close();}

下载成功！

小文件合并

由于HDFS采用块机制，再小的文件也会占用一个块，会浪费资源，某些场合可以将若干小文件合并上传至HDFS，来减少内存浪费!

@Testpublic void mergeFile() throws URISyntaxException, IOException, InterruptedException {//获取分布式文件系统FileSystem fileSystem = FileSystem.get(new URI("hdfs://node01:8020"), new Configuration(),"root");FSDataOutputStream outputStream = fileSystem.create(newPath("/bigfile.txt"));//获取本地文件系统LocalFileSystem local = FileSystem.getLocal(newConfiguration());//通过本地文件系统获取文件列表，为一个集合FileStatus[] fileStatuses = local.listStatus(newPath("E:\input"));for(FileStatus fileStatus : fileStatuses) { FSDataInputStream inputStream = local.open(fileStatus.getPath()); IOUtils.copyBytes(inputStream,outputStream,1024); IOUtils.closeStream(inputStream); }IOUtils.closeStream(outputStream);local.close();fileSystem.close();}

以上就是HDFS的基本API操作。

HDFS的高可用机制

在Hadoop中，整个HDFS文件系统的元数据信息都由NameNode 来管理，NameNode的可用性直接决定了Hadoop 的可用性，一旦NameNode进程不能工作了，就会影响整个集群的正常使用。

在典型的高可用(HA)集群中，两台独立的机器被配置为NameNode。在工作集群中，NameNode机器中的一个处于Active状态，另一个处于Standby状态。Active NameNode负责群集中的所有客户端操作，而Standby NameNode充当从服务器。Standby机器保持足够的状态以提供快速故障切换。

NameNode包含了HDFS的元数据信息和数据块信息（blockmap），其中数据块信息通过DataNode主动向Active NameNode和Standby NameNode上报。

共享存储系统负责存储HDFS的元数据（EditsLog），Active NameNode的写入和 Standby NameNode的读取，通过共享存储系统实现元数据同步。在主备切换过程中，新的Active NameNode必须确保元数据同步完成才能对外提供服务。

ZKFC是一个进程，由三个组件组成：

ZKFailoverController(ZKFC)：是基于Zookeeper的故障转移控制器，负责控制NameNode的主备切换，ZKFailoverController会监测NameNode的健康状态，当发现Active NameNode出现异常时会通过Zookeeper进行一次新的选举，完成Active和Standby状态的切换。

HealthMonitor(HM):周期性调用NameNode的HAServiceProtocol RPC接口（monitorHealth 和 getServiceStatus），简单来说就是心跳机制，监控NameNode的健康状态并向ZKFC组件反馈。

ActiveStandbyElector(ASE)：接收ZKFC组件的选举请求，通过Zookeeper自动完成主备选举，选举完成后回调ZKFC组件的主备切换方法，对NameNode进行Active和Standby状态的切换。

Hadoop的联邦机制

单NameNode的架构使得HDFS在集群扩展性和性能上都有潜在的问题，当集群大到一定程度后，NameNode进程使用的内存可能会达到上百G，NameNode成为了性能的瓶颈，因而提出了NameNode水平扩展方案-- Federation。

Federation中文意思为联邦，是NameNode的Federation,也就是会有多个NameNode，这些namenode之间是联合的，它们之间相互独立且不需要互相协调，各自分工，管理自己的区域，共享集群中所有的DataNode的,它们还是在同一个集群内的。

分布式的datanode被用作通用的数据块存储存储设备。每个datanode要向集群中所有的namenode注册，且周期性地向所有namenode发送心跳和块报告，并执行来自所有namenode的命令。DataNode上不仅仅存储一个Block Pool下的数据,而是多个Block Pool下的数据。

多个NN共用一个集群里的存储资源，每个NN都可以单独对外提供服务。

每个NN都会定义一个存储池，有单独的id，每个DN都为所有存储池提供存储。DN会按照存储池id向其对应的NN汇报块信息，同时，DN会向所有NN汇报本地存储可用资源情况。

HDFS Federation不足HDFS Federation并没有完全解决单点故障问题。虽然namenode/namespace存在多个，但是从单个namenode/namespace看，仍然存在单点故障问题。如果某个namenode挂掉了，其管理的相应的文件便不可以访问。

Federation中每个namenode配有一个secondary namenode，便于在主namenode挂掉时还原元数据信息。所以一般集群规模真的很大的时候，会采用HA+Federation的部署方案，每个联合的namenodes都是HA的。