Hadoop大数据通用处理平台

时间  2019-11-05
标签 hadoop 数据 通用 处理 平台 栏目 Hadoop 繁體版
原文   原文链接

1.简介node

Hadoop是一款开源的大数据通用处理平台，其提供了分布式存储和分布式离线计算，适合大规模数据、流式数据(写一次,读屡次)，不适合低延时的访问、大量的小文件以及频繁修改的文件。数据库

*Hadoop由HDFS、YARN、MapReduce组成。服务器

Hadoop的特色：网络

1.高扩展(动态扩容)：可以存储和处理千兆字节数据(PB)，可以动态的增长和卸载节点，提高存储能力(可以达到上千个节点)框架

2.低成本：只须要普通的PC机就能实现，不依赖高端存储设备和服务器。ssh

3.高效率：经过在Hadoop集群中分化数据并行处理，使得处理速度很是快。分布式

4.可靠性：数据有多份副本，而且在任务失败后能自动从新部署。ide

Hadoop的使用场景：oop

1.日志分析，将数据分片并行计算处理。大数据

2.基于海量数据的在线应用。

3.推荐系统，精准营销。

4.搜索引擎。

Hadoop生态圈：

Hive：利用Hive能够不须要编写复杂的Hadoop程序，只须要写一个SQL语句，Hive就会把SQL语句转换成Hadoop的任务去执行，下降使用Hadoop离线计算的门槛。

HBase：海量数据存储的非关系型数据库，单个表中的数据可以容纳百亿行x百万列。

ZooKeeper：监控Hadoop集群中每一个节点的状态，管理整个集群的配置，维护节点间数据的一致性。

Flume：海量日志采集系统。

2.内部结构

2.1 HDFS

HDFS是分布式文件系统，存储海量的文件，其中HDFS中包含NameNode、DataNode、SecondaryNameNode组件等。

Block数据块

1.HDFS中基本的存储单元，1.X版本中每一个Block默认是64M，2.X版本中每一个Block默认是128M。

2.一个大文件会被拆分红多个Block进行存储，若是一个文件少于Block的大小，那么其实际占用的空间为文件自身大小。

3.每一个Block都会在不一样的DataNode节点中存在备份(默认备份数是3)

DataNode

1.保存具体的Blocks数据。

2.负责数据的读写操做和复制操做。

3.DataNode启动时会向NameNode汇报当前存储的数据块信息。

NameNode

1.存储文件的元信息和文件与Block、DataNode的关系，NameNode运行时全部数据都保存在内存中，所以整个HDFS可存储的文件数受限于NameNode的内存大小。

2.每一个Block在NameNode中都对应一条记录，若是是大量的小文件将会消耗大量内存，所以HDFS适合存储大文件。

3.NameNode中的数据会定时保存到本地磁盘中(只有元数据)，但不保存文件与Block、DataNode的位置信息，这部分数据由DataNode启动时上报和运行时维护。

*NameNode不容许DataNode具备同一个Block的多个副本，因此建立的最大副本数量是当时DataNode的总数。

*DataNode会按期向NameNode发送心跳信息，一旦在必定时间内NameNode没有接收到DataNode发送的心跳则认为其已经宕机，所以不会再给它任何IO请求。

*若是DataNode失效形成副本数量降低而且低于预先设置的阈值或者动态增长副本数量，则NameNode会在合适的时机从新调度DataNode进行复制。

SecondaryNameNode

1.定时与NameNode进行同步，合并HDFS中系统镜像，定时替换NameNode中的镜像。

HDFS写入文件的流程

1.HDFS Client向NameNode申请写入文件。

2.NameNode根据文件大小，返回文件要写入的DataNode列表以及Block id (此时NameNode已存储文件的元信息、文件与DataNode、Block之间的关系)

3.HDFS Client收到响应后，将文件写入第一个DataNode中，第一个DataNode接收到数据后将其写入本地磁盘，同时把数据传递给第二个DataNode，直到写入备份数个DataNode。

4.每一个DataNode接收完数据后都会向前一个DataNode返回写入成功的响应，最终第一个DataNode将返回HDFS Client客户端写入成功的响应。

5.当HDFS Client接收到整个DataNodes的确认请求后会向NameNode发送最终确认请求，此时NameNode才会提交文件。

*当写入某个DataNode失败时，数据会继续写入其余的DataNode，NameNode会从新寻找DataNode继续复制，以保证数据的可靠性。

*每一个Block都会有一个校验码并存放在独立的文件中，以便读的时候来验证数据的完整性。

*文件写入完毕后，向NameNode发送确认请求，此时文件才可见，若是发送确认请求以前NameNode宕机，那么文件将会丢失，HDFS客户端没法进行读取。

HDFS读取文件的流程

1.HDFS Client向NameNode申请读取指定文件。

2.NameNode返回文件全部的Block以及这些Block所在的DataNodes中(包括复制节点)

3.HDFS Client根据NameNode的返回，优先从与HDFS Client同节点的DataNode中直接读取(若HDFS Client不在集群范围内则随机选择)，若是从DataNode中读取失败则经过网络从复制节点中进行读取。

机架感知

分布式集群中一般包含很是多的机器，因为受到机架槽位和交换机网口的限制，一般大型的分布式集群都会跨好几个机架，由多个机架上的机器共同组成一个分布式集群。

机架内的机器之间的网络速度一般都会高于跨机架机器之间的网络速度，而且机架之间机器的网络通讯一般受到上层交换机间网络带宽的限制。

Hadoop默认没有开启机架感知功能，默认状况下每一个Block都是随机分配DataNode，须要进行相关的配置，那么在NameNode启动时，会将机器与机架的对应信息保存在内存中，用于在HDFS Client申请写文件时，可以根据预先定义的机架关系合理的分配DataNode。

Hadoop机架感知默认对3个副本的存放策略为：

第1个Block副本存放在和HDFS Client所在的节点中(若HDFS Client不在集群范围内则随机选取)

第2个Block副本存放在与第一个节点不一样机架下的节点中(随机选择)

第3个Block副本存放在与第2个副本所在节点的机架下的另外一个节点中，若是还有更多的副本则随机存放在集群的节点中。

*使用此策略能够保证对文件的访问可以优先在本机架下找到，而且若是整个机架上发生了异常也能够在另外的机架上找到该Block的副本。

2.2 YARN

YARN是分布式资源调度框架(任务计算框架的资源调度框架)，主要负责集群中的资源管理以及任务调度而且监控各个节点。

ResourceManager

1.是整个集群的资源管理者，管理并监控各个NodeManager。

2.处理客户端的任务请求。

3.启动和监控ApplicationMaster。

4.负责资源的分配以及调度。

NodeManager

1.是每一个节点的管理者，负责任务的执行。

2.处理来自ResourceManager的命令。

3.处理来自ApplicationMaster的命令。

ApplicationMaster

1.数据切分，用于并行计算处理。

2.计算任务所须要的资源。

3.负责任务的监控与容错。

任务运行在YARN的流程

1.客户端提交任务请求到ResourceManager。

2.ResourceManager生成一个ApplicationManager进程，用于任务的管理。

3.ApplicationManager建立一个Container容器用于存听任务所须要的资源。

4.ApplicationManager寻找其中一个NodeManager，在此NodeManager中启动一个ApplicationMaster，用于任务的管理以及监控。

5.ApplicationMaster向ResourceManager进行注册，并计算任务所需的资源汇报给ResourceManager(CPU与内存)

6.ResourceManager为此任务分配资源，资源封装在Container容器中。

7.ApplicationMaster通知集群中相关的NodeManager进行任务的执行。

8.各个NodeManager从Container容器中获取资源并执行Map、Reduce任务。

2.3 MapReduce

MapReduce是分布式离线并行计算框架，高吞吐量，高延时，原理是将分析的数据拆分红多份，经过多台节点并行处理，相对于Storm、Spark任务计算框架而言，MapReduce是最先出现的计算框架。

MapReduce、Storm、Spark任务计算框架对比：

MapReduce执行流程

MapReduce将程序划分为Map任务以及Reduce任务两部分。

Map任务处理流程

1.读取文件中的内容，解析成Key-Value的形式 (Key为偏移量，Value为每行的数据)

2.重写map方法，编写业务逻辑，生成新的Key和Value。

3.对输出的Key、Value进行分区(Partitioner类)

4.对数据按照Key进行排序、分组，相同key的value放到一个集合中(数据汇总)

*处理的文件必需要在HDFS中。

Reduce任务处理流程

1.对多个Map任务的输出，按照不一样的分区，经过网络复制到不一样的reduce节点。

2.对多个Map任务的输出进行合并、排序。

3.将reduce的输出保存到文件，存放在HDFS中。

3.Hadoop的使用

3.1 安装

因为Hadoop使用Java语言进行编写，所以须要安装JDK。

从CDH中下载Hadoop 2.X并进行解压，CDH是Cloudrea公司对各类开源框架的整合与优化(较稳定)

etc目录：Hadoop配置文件存放目录。

logs目录：Hadoop日志存放目录。

bin目录、sbin目录：Hadoop可执行命令存放目录。

etc目录

bin目录

sbin目录

3.2 Hadoop配置

1.配置环境

编辑etc/hadoop/hadoop-env.sh文件，修改JAVA_HOME配置项为本地JAVA的HOME目录，此文件是Hadoop启动时加载的环境变量。

编辑/etc/hosts文件，添加主机名与IP的映射关系。

2.配置Hadoop公共属性（core-site.xml）

复制代码
<configuration>
<!-- Hadoop工做目录,用于存放Hadoop运行时产生的临时数据 -->
<property>
<name>hadoop.tmp.dir</name>
<value>/usr/hadoop/hadoop-2.9.0/data</value>
</property>
<!-- NameNode的通讯地址,1.x默认9000,2.x可使用8020 -->
<property>
<name>fs.default.name</name>
<value>hdfs://192.168.1.80:8020</value>
</property>
</configuration>
复制代码

3.配置HDFS（hdfs-site.xml）

复制代码
<configuration>
<!--指定block的副本数量(将block复制到集群中备份数-1个节点的DataNode中)-->
<property>
<name>dfs.replication</name>
<value>1</value>
</property>
<!-- 关闭HDFS的访问权限 -->
<property>
<name>dfs.permissions.enabled</name>
<value>false</value>
</property>
</configuration>
复制代码

4.配置YARN（yarn-site.xml）

复制代码
<configuration>
<!-- 配置Reduce取数据的方式是shuffle(随机) -->
<property>
<name>yarn.nodemanager.aux-services</name>
<value>mapreduce_shuffle</value>
</property>
</configuration>
复制代码

5.配置MapReduce（mapred-site.xml）

复制代码
<configuration>
<!-- 让MapReduce任务使用YARN进行调度 -->
<property>
<name>mapreduce.framework.name</name>
<value>yarn</value>
</property>
</configuration>
复制代码

6.配置SSH

因为在启动hdfs、yarn时都须要对用户的身份进行验证，所以能够配置SSH设置免密码登陆。

//生成秘钥
ssh-keygen -t rsa

//复制秘钥到本机
ssh-copy-id 192.168.1.80

3.3 启动HDFS

1.格式化NameNode

2.启动HDFS，将会启动NameNode、DataNode、SecondaryNameNode三个进程，能够经过jps命令进行查看。

*若启动时出现错误，则能够进入logs目录查看相应的日志文件。

当HDFS启动完毕后，能够访问http://localhost:50070进入HDFS的可视化管理界面，能够在此页面中监控整个HDFS集群的情况而且进行文件的上传以及下载。

*进入HDFS监控页面下载文件时，会将请求重定向，重定向后的地址的主机名为NameNode的主机名，所以客户端本地的host文件中须要配置NameNode主机名与IP的映射关系。

3.4 启动YARN

启动YARN后，将会启动ResourceManager以及NodeManager进程，能够经过jps命令进行查看。

当YARN启动完毕后，能够访问http://localhost:8088进入YARN的可视化管理界面，能够在此页面中查看任务的执行状况以及资源的分配。

3.5 使用Shell命令操做HDFS

HDFS中的文件系统与Linux相似，由/表明根目录。

复制代码
hadoop fs -cat <src>：显示文件中的内容。

hadoop fs -copyFromLocal <localsrc> <dst>：将本地中的文件上传到HDFS。

hadoop fs -copyToLocal <src> <localdst>：将HDFS中的文件下载到本地。

hadoop fs -count <path>：查询指定路径下文件的个数。

hadoop fs -cp <src> <dst>：在HDFS内对文件进行复制。

hadoop fs -get <src> <localdst>：将HDFS中的文件下载到本地。

hadoop fs -ls <path>：显示指定目录下的内容。

hadoop fs -mkdir <path>：建立目录。

hadoop fs -moveFromLocal <localsrc> <dst>：将本地中的文件剪切到HDFS中。

hadoop fs -moveToLocal <src> <localdst> :将HDFS中的文件剪切到本地中。

hadoop fs -mv <src> <dst> :在HDFS内对文件进行移动。

hadoop fs -put <localsrc> <dst>：将本地中的文件上传到HDFS。

hadoop fs -rm <src>：删除HDFS中的文件。
复制代码

3.6 JAVA中操做HDFS

复制代码
/**

  • @Auther: ZHUANGHAOTANG
  • @Date: 2018/11/6 11:49

  • @Description:
    */
    public class HDFSUtils {

    /**

    • HDFS NamenNode URL
      */
      private static final String NAMENODE_URL = "hdfs://hadoop1:8020";

    /**

    • 配置项
      */
      private static Configuration conf = null;

    static {
    conf = new Configuration();
    }

    /**

    • 建立目录
      */
      public static void mkdir(String dir) throws Exception {
      if (StringUtils.isBlank(dir)) {
      throw new Exception("Parameter Is NULL");
      }
      dir = NAMENODE_URL + dir;
      FileSystem fs = FileSystem.get(URI.create(NAMENODE_URL), conf);
      if (!fs.exists(new Path(dir))) {
      fs.mkdirs(new Path(dir));
      }
      fs.close();
      }

    /**

    • 删除目录或文件
      */
      public static void delete(String dir) throws Exception {
      if (StringUtils.isBlank(dir)) {
      throw new Exception("Parameter Is NULL");
      }
      dir = NAMENODE_URL + dir;
      FileSystem fs = FileSystem.get(URI.create(NAMENODE_URL), conf);
      fs.delete(new Path(dir), true);
      fs.close();
      }

    /**

    • 遍历指定路径下的目录和文件
      */
      public static List<String> listAll(String dir) throws Exception {
      List<String> names = new ArrayList<>();
      if (StringUtils.isBlank(dir)) {
      throw new Exception("Parameter Is NULL");
      }
      dir = NAMENODE_URL + dir;
      FileSystem fs = FileSystem.get(URI.create(dir), conf);
      FileStatus[] files = fs.listStatus(new Path(dir));
      for (int i = 0, len = files.length; i < len; i++) {
      if (files[i].isFile()) { //文件
      names.add(files[i].getPath().toString());
      } else if (files[i].isDirectory()) { //目录
      names.add(files[i].getPath().toString());
      } else if (files[i].isSymlink()) { //软或硬连接
      names.add(files[i].getPath().toString());
      }
      }
      fs.close();
      return names;
      }

    /**

    • 上传当前服务器的文件到HDFS中
      */
      public static void uploadLocalFileToHDFS(String localFile, String hdfsFile) throws Exception {
      if (StringUtils.isBlank(localFile) || StringUtils.isBlank(hdfsFile)) {
      throw new Exception("Parameter Is NULL");
      }
      hdfsFile = NAMENODE_URL + hdfsFile;
      FileSystem fs = FileSystem.get(URI.create(NAMENODE_URL), conf);
      Path src = new Path(localFile);
      Path dst = new Path(hdfsFile);
      fs.copyFromLocalFile(src, dst);
      fs.close();
      }

    /**

    • 经过流上传文件
      */
      public static void uploadFile(String hdfsPath, InputStream inputStream) throws Exception {
      if (StringUtils.isBlank(hdfsPath)) {
      throw new Exception("Parameter Is NULL");
      }
      hdfsPath = NAMENODE_URL + hdfsPath;
      FileSystem fs = FileSystem.get(URI.create(NAMENODE_URL), conf);
      FSDataOutputStream os = fs.create(new Path(hdfsPath));
      BufferedInputStream bufferedInputStream = new BufferedInputStream(inputStream);
      byte[] data = new byte[1024];
      while (bufferedInputStream.read(data) != -1) {
      os.write(data);
      }
      os.close();
      fs.close();
      }

    /**

    • 从HDFS中下载文件
      */
      public static byte[] readFile(String hdfsFile) throws Exception {
      if (StringUtils.isBlank(hdfsFile)) {
      throw new Exception("Parameter Is NULL");
      }
      hdfsFile = NAMENODE_URL + hdfsFile;
      FileSystem fs = FileSystem.get(URI.create(NAMENODE_URL), conf);
      Path path = new Path(hdfsFile);
      if (fs.exists(path)) {
      FSDataInputStream is = fs.open(path);
      FileStatus stat = fs.getFileStatus(path);
      byte[] data = new byte[(int) stat.getLen()];
      is.readFully(0, data);
      is.close();
      fs.close();
      return data;
      } else {
      throw new Exception("File Not Found In HDFS");
      }
      }

}
复制代码

3.7 执行一个MapReduce任务

Hadoop中提供了hadoop-mapreduce-examples-2.9.0.jar，其封装了一些任务计算方法，能够直接进行调用。

*使用hadoop jar命令执行JAR包。

1.建立一个文件，将此文件上传到HDFS中。

2.使用Hadoop提供的hadoop-mapreduce-examples-2.9.0.jar执行wordcount词频统计功能，而后在YARN管理页面中进行查看。

YARN管理页面中能够查看任务的执行进度：

3.当任务执行完毕后，能够查看任务的执行结果。

*任务的执行结果将会放到HDFS的文件中。

相关文章
相关标签/搜索
每日一句
    每一个你不满意的现在，都有一个你没有努力的曾经。
本站公众号
   欢迎关注本站公众号,获取更多信息
联系我们 最近搜索 最新文章 沪ICP备13005482号-10 MyBatis教程 SQL 教程 MySQL教程 Java 教程 Thymeleaf 教程 Hibernate教程 Spring教程 Redis教程