大数据技术原理与应用 - (8). Hive - 基于 Hadoop 的数据仓库

【第三篇】 - 大数据处理与分析, 《大数据技术原理与应用, 林子雨》

本篇介绍大数据处理与分析的相关技术，包括

• 第7章 - MapReduce
• 第8章 - Hive - 基于 Hadoop 的数据仓库
• 第9章 - Hadoop 的优化与发展
• 第10章 - Spark
• 第11章 - 流计算
• 第12章 - 图计算

Hive 是一个基于 Hadoop 的数据仓库平台。通过 Hive，我们可以方便地进行 ETL 的工作。Hive 定义了一个类似于 SQL 的查询语言: HiveQL，能够将用户编写的 HiveQL 转化为相应的 Mapreduce 程序基于 Hadoop 执行，可以说 Hive 实质就是一款基于 HDFS 的 MapReduce 计算框架，对存储在 HDFS 中的数据进行分析和管理。

Hive 概述

数据仓库概念

数据仓库 (Data Warehouse) 是一个面向主题的 (Subject Oriented)、集成的 (Integrated)、相对稳定的 (Non-Volatile)、反映历史变化 (Time Variant) 的数据集合，用于支持管理决策。

传统数据仓库面临的挑战

1. 无法满足快速增长的海量数据存储需求
2. 无法有效处理不同类型的数据
3. 计算和处理能力不足

Hive 简介

• Hive 是一个构建于 Hadoop 顶层的数据仓库工具
• 支持大规模数据存储、分析，具有良好的可扩展性
• 某种程度上可以看作是用户编程接口，本身不存储和处理数据
• 依赖分布式文件系统 HDFS 存储数据
• 依赖分布式并行计算模型 MapReduce 处理数据
• 定义了简单的类似 SQL 的查询语言 —— HiveQL
• 用户可以通过编写的 HiveQL 语句运行 MapReduce 任务
• 可以很容易把原来构建在关系数据库上的数据仓库应用程序移植到 Hadoop 平台上
• 是一个可以提供有效、合理、直观组织和使用数据的分析工具

Hive 特点

Hive 具有的特点非常适用于数据仓库

1. 采用批处理方式处理海量数据
   • Hive 需要把 HiveQL 语句转换成 MapReduce 任务进行运行
   • 数据仓库存储的是静态数据，对静态数据的分析适合采用批处理方式，不需要快速响应给出结果，而且数据本身也不会频繁变化
2. 提供适合数据仓库操作的工具
   • Hive 本身提供了一系列对数据进行提取、转换、加载 (ETL) 的工具，可以存储、查询和分析存储在 Hadoop 中的大规模数据
   • 这些工具能够很好地满足数据仓库各种应用场景

• 优点：
  1. 可扩展性,横向扩展，Hive 可以自由的扩展集群的规模，一般情况下不需要重启服务 横向扩展：通过分担压力的方式扩展集群的规模 纵向扩展：一台服务器 cpu i7-6700k 4 核心 8 线程，8 核心 16 线程，内存 64G => 128G
  2. 延展性，Hive 支持自定义函数，用户可以根据自己的需求来实现自己的函数
  3. 良好的容错性，可以保障即使有节点出现问题，SQL 语句仍可完成执行
• 缺点：
  1. Hive 不支持记录级别的增删改操作，但是用户可以通过查询生成新表或者将查询结 果导入到文件中 (当前选择的 hive-2.3.2 的版本支持记录级别的插入操作)
  2. Hive 的查询延时很严重，因为 MapReduce Job 的启动过程消耗很长时间，所以不能 用在交互查询系统中。
  3. Hive 不支持事务 (因为不没有增删改，所以主要用来做 OLAP (联机分析处理)，而不是 OLTP (联机事务处理)，这就是数据处理的两大级别)。

Hive 和 RDBMS 的对比

总结：

• Hive 具有 SQL 数据库的外表，但应用场景完全不同，Hive 只适合用来做海量离线数 据统计分析，也就是数据仓库。

Hive在企业中的部署和应用

• Hive 批处理，时延高，报表不需要实时出结果。
• HBase 可以实时反馈。
• Mahout 中实现和提供了很多机器学习的 api，帮助企业分析人员能够快速构建商业智能应用。
• HDFS：分布式文件存储；MapReduce：分布式数据处理

Hive Design

Hive Architecture

• 用户接口 (UI) 模块包括 CLI、HWI、JDBC、ODBC、Thrift Server
• 驱动模块 (Driver) 包括 Compiler、Optimizer、Execution Engine 等，负责把 HiveSQL 语句转换成一系列 MapReduce 作业
• 元数据 (metadata) 存储模块 (Metastore) 是一个独立的关系型数据库 (自带 derby 数据库，或 MySQL 数据库)

• Source
• 中文翻译

Apache Hive wiki: https://cwiki.apache.org/confluence/display/Hive/Design

Figure above shows the major components of Hive and its interactions with Hadoop. As shown in that figure, the main components of Hive are:

• UI – The user interface for users to submit queries and other operations to the system. As of 2011 the system had a command line interface and a web based GUI was being developed.
• Driver – The component which receives the queries. This component implements the notion of session handles and provides execute and fetch APIs modeled on JDBC/ODBC interfaces.
• Compiler – The component that parses the query, does semantic analysis on the different query blocks and query expressions and eventually generates an execution plan with the help of the table and partition metadata looked up from the metastore.
• Metastore – The component that stores all the structure information of the various tables and partitions in the warehouse including column and column type information, the serializers and deserializers necessary to read and write data and the corresponding HDFS files where the data is stored.
• Execution Engine – The component which executes the execution plan created by the compiler. The plan is a DAG of stages. The execution engine manages the dependencies between these different stages of the plan and executes these stages on the appropriate system components.

Figure also shows how a typical query flows through the system. The UI calls the execute interface to the Driver (step 1 in Figure 1). The Driver creates a session handle for the query and sends the query to the compiler to generate an execution plan (step 2). The compiler gets the necessary metadata from the metastore (steps 3 and 4). This metadata is used to typecheck the expressions in the query tree as well as to prune partitions based on query predicates. The plan generated by the compiler (step 5) is a DAG of stages with each stage being either a map/reduce job, a metadata operation or an operation on HDFS. For map/reduce stages, the plan contains map operator trees (operator trees that are executed on the mappers) and a reduce operator tree (for operations that need reducers). The execution engine submits these stages to appropriate components (steps 6, 6.1, 6.2 and 6.3). In each task (mapper/reducer) the deserializer associated with the table or intermediate outputs is used to read the rows from HDFS files and these are passed through the associated operator tree. Once the output is generated, it is written to a temporary HDFS file though the serializer (this happens in the mapper in case the operation does not need a reduce). The temporary files are used to provide data to subsequent map/reduce stages of the plan. For DML operations the final temporary file is moved to the table‘s location. This scheme is used to ensure that dirty data is not read (file rename being an atomic operation in HDFS). For queries, the contents of the temporary file are read by the execution engine directly from HDFS as part of the fetch call from the Driver (steps 7, 8 and 9).

上图展示了 Hive 的的主要组件以及它们与 Hadoop 的交互过程，这些 Hive 的主要组件包括:

• UI - 用户界面。用于向系统提交查询和其他操作。截至2011年，该系统具有命令行界面，并且正在开发基于Web的GUI。
• Driver - 接收查询的组件。该组件实现了 session handle 的概念，并提供了以 JDBC / ODBC 接口为模型的 API 的执行和获取。
• Compiler - 解析查询的组件。对不同的查询块和查询表达式进行语义分析，并最终借助 table 和从 metastore 查找的 partition metadata (分区元数据) 来生成执行计划。
• Metastore - 该组件用于存储所有在仓库中的各种的 tables 和 partitons 的结构信息，包括列和列类型信息，读写数据所需的 serializers 和 deserializers 以及存储数据的相应 HDFS 文件。
• Execution Engine - 执行由 Compiler 创建的执行计划的组件。该计划是阶段的 DAG。Execution Engine 管理 execution plan 不同阶段之间的 dependencies，并在适当的系统组件上执行这些阶段。

该图还展示了一个典型的查询如何在系统中 flow。UI调用调用 Driver 的执行接口（步骤1）。Driver 为 query 创建 session handle，并将 query 发送到 Compiler 以生成执行计划（步骤2）。Compiler 从 metastore 中获取必要的 metadata（步骤3和4）。该 metadata 用于对查询树中 (query tree) 的表达式进行类型检查，以及基于查询谓词 (query predicates) 修剪 partition。编译器生成的计划（步骤5）是阶段的DAG，每个阶段是一个 Map / Reduce 作业，metadata 操作或对 HDFS 的操作。对于 Map / Reduce阶段，该 plan 包含 Map Operator Trees（在 Mapper 上执行的 Operator Trees）和 Reduce Operator Tree（用于需要 Reducers 的运算）。Execution Engine 将这些 stages 将这些阶段提交给适当的组件（步骤6、6.1、6.2和6.3）。在每个任务（Mapper/ Reducer）中，与 table 或中间输出关联的 deserializer 用于从 HDFS 文件读取行，并将这些行通过关联的 operator tree 传递。生成输出后，将通过 serializer 将其写入临时 HDFS 文件中（如果不需要 Reduce 操作，则在 Napper 发生）。临时文件用于向计划的后续 Map/ Reduce 阶段提供数据。对于 DML 操作，最终的临时文件将移动到 table 的位置。此方案用于确保脏数据 (dirty data) 不会被读取（文件重命名是 HDFS 中的 atomic 操作）。对于查询 (query)，Execution Engine直接从 HDFS 读取临时文件的内容，作为 Driver 进行提取调用的一部分（步骤7、8和9）。

Hive Data Model

Data in Hive is organized into:

• Tables – These are analogous to Tables in Relational Databases. Tables can be filtered, projected, joined and unioned. Additionally all the data of a table is stored in a directory in HDFS. Hive also supports the notion of external tables wherein a table can be created on prexisting files or directories in HDFS by providing the appropriate location to the table creation DDL. The rows in a table are organized into typed columns similar to Relational Databases.
• Partitions – Each Table can have one or more partition keys which determine how the data is stored, for example a table T with a date partition column ds had files with data for a particular date stored in the <table location>/ds=<date> directory in HDFS. Partitions allow the system to prune data to be inspected based on query predicates, for example a query that is interested in rows from T that satisfy the predicate T.ds = ‘2008-09-01’ would only have to look at files in <table location>/ds=2008-09-01/ directory in HDFS.
• Buckets – Data in each partition may in turn be divided into Buckets based on the hash of a column in the table. Each bucket is stored as a file in the partition directory. Bucketing allows the system to efficiently evaluate queries that depend on a sample of data (these are queries that use the SAMPLE clause on the table).

Apart from primitive column types (integers, floating point numbers, generic strings, dates and booleans), Hive also supports arrays and maps. Additionally, users can compose their own types programmatically from any of the primitives, collections or other user-defined types. The typing system is closely tied to the SerDe (Serailization/Deserialization) and object inspector interfaces. User can create their own types by implementing their own object inspectors, and using these object inspectors they can create their own SerDes to serialize and deserialize their data into HDFS files). These two interfaces provide the necessary hooks to extend the capabilities of Hive when it comes to understanding other data formats and richer types. Builtin object inspectors like ListObjectInspector, StructObjectInspector and MapObjectInspector provide the necessary primitives to compose richer types in an extensible manner. For maps (associative arrays) and arrays useful builtin functions like size and index operators are provided. The dotted notation is used to navigate nested types, for example a.b.c = 1 looks at field c of field b of type a and compares that with 1.

Hive 工作原理

• SQL 语句转换成 MapReduce 作业的基本原理
• Hive 中 SQL 查询转换成 MapReduce 作业的过程

SQL 语句转换成 MapReduce 的基本原理

1. join 的实现原理

过程与第 7 章 —— 用 MapReduce 实现关系的自然连接类似，请参见上一章 MapReduce。

2. group by 的实现原理

存在一个分组 (Group By) 操作，其功能是把表 Score 的不同片段按照 rank 和 level 的组合值进行合并，计算不同 rank 和 level 的组合值分别有几条记录：

select rank, level ,count(*) as value from score group by rank, level

Hive 中 SQL 查询转换成 MapReduce 作业的过程

当用户向Hive输入一段命令或查询时，Hive需要与Hadoop交互工作来完成该操作:

• 驱动模块接收该命令或查询编译器
• 对该命令或查询进行解析编译
• 由优化器对该命令或查询进行优化计算
• 该命令或查询通过执行器进行执行

请参见上面 Sec 2.1 - Hive Architecture 介绍的图及过程。

1. 由 Hive 驱动模块中的编译器对用户输入的 SQL 语言进行词法和语法解析，将 SQL 语句转化为抽象语法树的形式
2. 抽象语法树的结构仍很复杂，不方便直接翻译为 MapReduce 算法程序，因此，把抽象语法书转化为查询块
3. 把查询块转换成逻辑查询计划，里面包含了许多逻辑操作符
4. 重写逻辑查询计划，进行优化，合并多余操作，减少 MapReduce 任务数量
5. 将逻辑操作符转换成需要执行的具体 MapReduce 任务
6. 对生成的 MapReduce 任务进行优化，生成最终的 MapReduce 任务执行计划
7. 由 Hive 驱动模块中的执行器，对最终的 MapReduce 任务进行执行输出

几点说明

• 当启动 MapReduce 程序时，Hive 本身是不会生成 MapReduce 算法程序的
• 需要通过一个表示 “Job 执行计划” 的 XML 文件驱动执行内置的、原生的 Mapper 和 Reducer 模块
• Hive 通过和 JobTracker 通信来初始化 MapReduce 任务，不必直接部署在 JobTracker 所在的管理节点上执行
• 通常在大型集群上，会有专门的网关机来部署 Hive 工具。网关机的作用主要是远程操作和管理节点上的 JobTracker 通信来执行任务
• 数据文件通常存储在 HDFS 上，HDFS 由名称节点管理

Hive High Availability (Hive HA)

• 将若干 Hive 实例纳入一个资源池，并由 HAProxy 提供一个统一的对外接口
• 对于程序开发人员，就把它认为是一台超强 “hive” 就可以。每次它接收到一个 Hive 查询连接后，都会轮询资源池里可用的 Hive 资源。这样，能充分使用每个 Hive server，减少压力。在拿到 Hive 连接后，Hive HA 会首先进行逻辑可用测试，这个逻辑规则可自行配置。
  • 如果逻辑可用，则直接把客户端的HIVE 查询连接 relay到该hive server。
  • 若逻辑不可用，则将该hiveserver放入黑名单，然后继续读取池里其他hive server进行连接测试。
• Hive HA 每隔一段时间 (可配置)，对黑名单中的 Hive server 进行处理，通过和节点管理服务器通讯，重启该 Hive server。如果重启后可用，则将该 Hive 从黑名单中移除，加入资源池。