走向云计算之Hive基本架构和使用详解

你的名字 2022-06-14 10:56 296阅读 0赞

## 一、概述 ##

### 1、什么是Hive ###

Hive是由facebook开源，最初用于解决海量结构化的日志数据统计问题的一种ETL（Extraction-Transformation-Loading）工具。它是构建在Hadoop之上的数据仓库，数据计算使用MapReduce，数据存储使用HDFS。Hive 定义了一种类 SQL 查询语言——HQL，类似SQL，但不完全相同。Hive通常用于进行离线数据处理（采用MapReduce）,可认为是一个HQL—>MR的语言翻译器。

### 2、为什么使用Hive ###

首先Hive较MapReduce而言，简单，容易上手，它提供了类SQL查询语言HQL。Hive可以应对超大数据集设计的计算/扩展能力。Hive以MR作为计算引擎，HDFS作为存储系统，有着统一的元数据管理（HCalalog）可与Pig、Presto等共享。

### 3、Hive和MapReduce ###

Hive不能代替MapReduce。理由如下：  
首先，Hive的HQL表达的能力有限

*  迭代式算法无法表达
 *  有些复杂运算用HQL不易表达

另外，Hive效率较低

*  Hive自动生成MapReduce作业，通常不够智能；
 *  HQL调优困难，粒度较粗
 *  可控性差

## 二、Hive的基本架构 ##

![这里写图片描述][20170618160030273]  
由上图可知，hadoop和mapreduce是hive架构的根基。Hive架构包括如下组件：CLI（command line interface）、JDBC/ODBC、Thrift Server、WEB GUI、metastore和Driver(Complier、Optimizer和Executor)，这些组件我可以分为两大类：服务端组件和客户端组件。

### 1、服务端组件： ###

*  Driver组件：该组件包括Complier、Optimizer和Executor，它的作用是将我们写的HiveQL（类SQL）语句进行解析、编译优化，生成执行计划，然后调用底层的mapreduce计算框架。
 *  Metastore组件：元数据服务组件，这个组件存储hive的元数据，hive的元数据存储在关系数据库里，hive支持的关系数据库有derby、mysql。元数据对于hive十分重要，因此hive支持把metastore服务独立出来，安装到远程的服务器集群里，从而解耦hive服务和metastore服务，保证hive运行的健壮性。
 *  Thrift服务：thrift是facebook开发的一个软件框架，它用来进行可扩展且跨语言的服务的开发，hive集成了该服务，能让不同的编程语言调用hive的接口。

### 2、客户端组件： ###

*  CLI：command line interface，命令行接口。
 *  Thrift客户端：上面的架构图里没有写上Thrift客户端，但是hive架构的许多客户端接口是建立在thrift客户端之上，包括JDBC和ODBC接口。
 *  WEBGUI：hive客户端提供了一种通过网页的方式访问hive所提供的服务。这个接口对应hive的hwi组件（hive web interface），使用前要启动hwi服务。

### 3、metastore组件详解 ###

*  Hive的metastore组件是hive元数据集中存放地。Metastore组件包括两个部分：metastore服务和后台数据的存储。后台数据存储的介质就是关系数据库，例如hive默认的嵌入式磁盘数据库derby，还有mysql数据库。
 *  Metastore服务是建立在后台数据存储介质之上，并且可以和hive服务进行交互的服务组件，默认情况下，metastore服务和hive服务是安装在一起的，运行在同一个进程当中。我也可以把metastore服务从hive服务里剥离出来，metastore独立安装在一个集群里，hive远程调用metastore服务，这样我们可以把元数据这一层放到防火墙之后，客户端访问hive服务，就可以连接到元数据这一层，从而提供了更好的管理性和安全保障。使用远程的metastore服务，可以让metastore服务和hive服务运行在不同的进程里，这样也保证了hive的稳定性，提升了hive服务的效率。
 *  部署实例  
    ![这里写图片描述][20170618160655066]

## 三、Hive Shell的基本使用 ##

### 1、启动Hadoop ###

我们知道，HDFS和Mapreduce是Hive架构的根基。因此，我们得先启动Hadoop，才能正确使用Hive。启动完毕后，直接输入命令hive即可进入hive shell中。注意在环境变量的path中添加hive的bin目录。  
![这里写图片描述][20170618162221737]

### 2、Hive的CLI命令行接口 ###

（1）内部表：与数据库中的 Table 在概念上是类似，每一个 Table 在 Hive 中都有一个相应的目录存储数据。例如，一个表 test，它在 HDFS 中的路径为：/ warehouse/test。 warehouse是在 hive-site.xml 中由 $\{hive.metastore.warehouse.dir\} 指定的数据仓库的目录。

*  创建表  
    `hive>CREATE TABLE user(id int);` // 创建内部表user，只有一个int类型的id字段  
    `hive>CREATE TABLE users(id int, name string);` // 创建内部表users，有两个字段。  
    ![这里写图片描述][20170618165138268]  
    `hive>CREATE TABLE users(id int, name string) ROW FORMAT DELIMITED FIELDS TERMINATED BY '\t';` // 创建内部表users，有两个字段，它们之间通过tab分隔。
 *  加载数据  
    创建一个user文件，内容如下：  
    ![这里写图片描述][20170618163917543]  
    `hive>LOAD DATA LOCAL INPATH '/usr/local/users' INTO TABLE users;` // 从本地文件加载  
    ![这里写图片描述][20170618164023263]  
    `hive>LOAD DATA INPATH '/root/id' INTO TABLE t1;` // 从HDFS中加载
 *  查看数据  
    `hive>select * from users;` // 该查询不启动的MapReduce  
    ![这里写图片描述][20170618165254581]
 *  `hive>select name from users` //该查询将启动MapReduce  
    ![这里写图片描述][20170618165540817]  
    ![这里写图片描述][20170618165552395]  
    `hive>select count(*) from users;` // hive也提供了聚合函数供使用，该查询将使用MapReduce。  
    ![这里写图片描述][20170618165734958]
 *  删除表  
    `hive>drop table users;`  
    ![这里写图片描述][20170618165825959]

（2）分区表：所谓分区（Partition） 对应于数据库的 Partition 列的密集索引。在 Hive 中，表中的一个 Partition 对应于表下的一个目录，所有的 Partition 的数据都存储在对应的目录中。例如：test表中包含 date 和 city 两个 Partition，则对应于date=20130201, city = bj 的 HDFS 子目录为：/warehouse/test/date=20130201/city=bj。而对应于date=20130202, city=sh 的HDFS 子目录为：/warehouse/test/date=20130202/city=sh。

*  创建表  
    `hive>CREATE TABLE users(id int,name string) PARTITIONED BY (day int);`
 *  加载表  
    `hive>LOAD DATA LOCAL INPATH '/usr/local/users' INTO TABLE users PARTITION (day=22);`

![这里写图片描述][20170618171443861]  
\- 查看创建情况  
退出hive使用命令`hdfs dfs -ls /user/hive/warehouse/users/`  
![这里写图片描述][20170618171608784]

（3）桶表（Hash 表）：桶表是对数据进行哈希取值，然后放到不同文件中存储。数据加载到桶表时，会对字段取hash值，然后与桶的数量取模。把数据放到对应的文件中。

*  创建表  
    `hive>CREATE TABLE users(id int,name string) clustered by(id) into 4 buckets;` // 创建一个桶表t4，根据id进行哈希取值，并设置4个桶来存储.
 *  设置允许进行分桶  
    `hive>set hive.enforce.bucketing = true;`
 *  插入数据  
    `hive>LOAD DATA LOCAL INPATH '/usr/local/users' INTO TABLE users;`

（4）外部表：它和内部表在元数据的组织上是相同的，而实际数据的存储则有较大的差异。外部表主要指向已经在 HDFS 中存在的数据，可以创建 Partition。  
在HDFS中的external目录下创建一个数据文件：`vim ids.txt`  
创建一个外部表：`hive>create external table t5(id int) location '/external';`  
外部表与内部表的差异：  
①内部表 的创建过程和数据加载过程（这两个过程可以在同一个语句中完成），在加载数据的过程中，实际数据会被移动到数据仓库目录中；之后对数据对访问将会直接在数据仓库目录中完成。删除表时，表中的数据和元数据将会被同时删除；  
②外部表 只有一个过程，加载数据和创建表同时完成，并不会移动到数据仓库目录中，只是与外部数据建立一个链接。当删除一个 外部表 时，仅删除该链接。

（5）视图操作：和关系数据库中的视图一个概念，可以向用户集中展现一些数据，屏蔽一些数据，提高数据库的安全性。

*  创建视图

hive> create view v1 as select * from t1;

*  查询视图

hive> select * from v1;

（6）查询操作：在Hive中，查询分为三种：基于Partition的查询、LIMIT Clause查询、Top N查询。  
①基于Partition的查询：一般 SELECT 查询是全表扫描。但如果是分区表，查询就可以利用分区剪枝（input pruning）的特性，类似“分区索引“”，只扫描一个表中它关心的那一部分。Hive 当前的实现是，只有分区断言（Partitioned by）出现在离 FROM 子句最近的那个WHERE 子句中，才会启用分区剪枝。例如，如果 page\_views 表（按天分区）使用 date 列分区，以下语句只会读取分区为‘2008-03-01’的数据。

SELECT page_views.*    FROM page_views    WHERE page_views.date >= '2013-03-01' AND page_views.date <= '2013-03-01'

②LIMIT Clause查询：Limit 可以限制查询的记录数。查询的结果是随机选择的。下面的查询语句从 t1 表中随机查询5条记录：

SELECT * FROM t1 LIMIT 5

③Top N查询：和关系型数据中的Top N一样，排序后取前N个显示。下面的查询语句查询销售记录最大的 5 个销售代表：

SET mapred.reduce.tasks = 1
    SELECT * FROM sales SORT BY amount DESC LIMIT 5

（7）连接操作：和关系型数据库中的各种表连接操作一样，在Hive中也可以进行表的内连接、外连接一类的操作：

*  导入ac信息表

hive> create table acinfo (name string,acip string)  row format delimited fields terminated by '\t' stored as TEXTFILE;
    hive> load data local inpath '/home/acinfo/ac.dat' into table acinfo;

*  内连接

select b.name,a.* from dim_ac a join acinfo b on (a.ac=b.acip) limit 10;

*  左外连接

select b.name,a.* from dim_ac a left outer join acinfo b on a.ac=b.acip limit 10;

## 四、Hive的Java API使用 ##

### 1、启动Hive的远程访问服务 ###

在服务器端启动Hive外部访问服务（不是在hive命令行模式下，而是普通模式下）：

hive --service hiveserver >/dev/null  2>/dev/null &

![这里写图片描述][20170618213335820]

### 2、给eclipse程序导入jar包 ###

在Hive的解压目录的lib包下找到Hive的相关jar包，导入eclipse中。  
![这里写图片描述][20170618213622835]

### 3、实例测试 ###

源码

package com.kang.hive;
    
    import java.sql.Connection;  
    import java.sql.DriverManager;  
    import java.sql.ResultSet;  
    import java.sql.SQLException;  
    import java.sql.Statement;  
    
    import org.apache.log4j.Logger;  
    
    /** 
     * Hive的JavaApi 
     *  
     * 运行该代码前请先启动hive的远程服务接口，直接在linux命令行下执行如下命令：
     * hive --service hiveserver >/dev/null  2>/dev/null &
     *  
     */  
    public class HiveApp {
          
    
        private static String driverName = "org.apache.hadoop.hive.jdbc.HiveDriver";  
        private static String url = "jdbc:hive://sparkproject1/default";  
        private static String user = "root";  
        private static String password = "root";  
        private static String sql = "";  
        private static ResultSet res;  
        private static final Logger log = Logger.getLogger(HiveApp.class);  
    
        public static void main(String[] args) {  
            Connection conn = null;  
            Statement stmt = null;  
            try {  
                conn = getConn();  
                stmt = conn.createStatement();  
    
                // 第一步:存在就先删除  
                String tableName = dropTable(stmt);  
    
                // 第二步:不存在就创建  
                createTable(stmt, tableName);  
    
                // 第三步:查看创建的表  
                showTables(stmt, tableName);  
    
                // 执行describe table操作  
                describeTables(stmt, tableName);  
    
                // 执行load data into table操作  
                loadData(stmt, tableName);  
    
                // 执行 select * query 操作  
                selectData(stmt, tableName);  
    
                // 执行 regular hive query 统计操作  
                countData(stmt, tableName);  
    
            } catch (ClassNotFoundException e) {  
                e.printStackTrace();  
                log.error(driverName + " not found!", e);  
                System.exit(1);  
            } catch (SQLException e) {  
                e.printStackTrace();  
                log.error("Connection error!", e);  
                System.exit(1);  
            } finally {  
                try {  
                    if (conn != null) {  
                        conn.close();  
                        conn = null;  
                    }  
                    if (stmt != null) {  
                        stmt.close();  
                        stmt = null;  
                    }  
                } catch (SQLException e) {  
                    e.printStackTrace();  
                }  
            }  
        }  
    
        private static void countData(Statement stmt, String tableName)  
                throws SQLException {  
            sql = "select count(1) from " + tableName;  
            System.out.println("Running:" + sql);  
            res = stmt.executeQuery(sql);  
            System.out.println("执行“regular hive query”运行结果:");  
            while (res.next()) {  
                System.out.println("count ------>" + res.getString(1));  
            }  
        }  
    
        private static void selectData(Statement stmt, String tableName)  
                throws SQLException {  
            sql = "select * from " + tableName;  
            System.out.println("Running:" + sql);  
            res = stmt.executeQuery(sql);  
            System.out.println("执行 select * query 运行结果:");  
            while (res.next()) {  
                System.out.println(res.getInt(1) + "\t" + res.getString(2));  
            }  
        }  
    
        private static void loadData(Statement stmt, String tableName)  
                throws SQLException {  
            String filepath = "/usr/local/users";  
            sql = "load data local inpath '" + filepath + "' into table "  
                    + tableName;  
            System.out.println("Running:" + sql);  
            res = stmt.executeQuery(sql);  
        }  
    
        private static void describeTables(Statement stmt, String tableName)  
                throws SQLException {  
            sql = "describe " + tableName;  
            System.out.println("Running:" + sql);  
            res = stmt.executeQuery(sql);  
            System.out.println("执行 describe table 运行结果:");  
            while (res.next()) {  
                System.out.println(res.getString(1) + "\t" + res.getString(2));  
            }  
        }  
    
        private static void showTables(Statement stmt, String tableName)  
                throws SQLException {  
            sql = "show tables '" + tableName + "'";  
            System.out.println("Running:" + sql);  
            res = stmt.executeQuery(sql);  
            System.out.println("执行 show tables 运行结果:");  
            if (res.next()) {  
                System.out.println(res.getString(1));  
            }  
        }  
    
        private static void createTable(Statement stmt, String tableName)  
                throws SQLException {  
            sql = "create table "  
                    + tableName  
                    + " (key int, value string)";  
            stmt.executeQuery(sql);  
        }  
    
        private static String dropTable(Statement stmt) throws SQLException {  
            // 创建的表名  
            String tableName = "testHive";  
            sql = "drop table " + tableName;  
            stmt.executeQuery(sql);  
            return tableName;  
        }  
    
        private static Connection getConn() throws ClassNotFoundException,  
                SQLException {  
            Class.forName(driverName);  
            Connection conn = DriverManager.getConnection(url, user, password);  
            return conn;  
        }  
    
    }

运行上面代码，结果如下：  
![这里写图片描述][20170618213929172]

对于最后一个查询数据数目时，Hive会启动一个MapReduce，所以运行时间较长。  
通过Hive Shell查询如下：  
![这里写图片描述][20170618214333225]

## 五、Hive的优化 ##

### 1、Hive性能低下的原因 ###

hive性能优化时，把HiveQL当做M/R程序来读，即从M/R的运行角度来考虑优化性能，从更底层思考如何优化运算性能，而不仅仅局限于逻辑代码的替换层面。Hadoop的核心能力是parition和sort，因而这也是优化的根本。观察Hadoop处理数据的过程，有几个显著的特征：

*  数据的大规模并不是负载重点，造成运行压力过大是因为运行数据的倾斜。
 *  jobs数比较多的作业运行效率相对比较低，比如即使有几百行的表，如果多次关联对此汇总，产生几十个jobs，将会需要30分钟以上的时间且大部分时间被用于作业分配，初始化和数据输出。M/R作业初始化的时间是比较耗时间资源的一个部分。
 *  在使用SUM，COUNT，MAX，MIN等UDAF函数时，不怕数据倾斜问题，Hadoop在Map端的汇总合并优化过，使数据倾斜不成问题。
 *  COUNT(DISTINCT)在数据量大的情况下，效率较低，如果多COUNT(DISTINCT)效率更低，因为COUNT(DISTINCT)是按GROUP BY字段分组，按DISTINCT字段排序，一般这种分布式方式是很倾斜的；比如：男UV，女UV，淘宝一天30亿的PV，如果按性别分组，分配2个reduce,每个reduce处理15亿数据。
 *  数据倾斜是导致效率大幅降低的主要原因，可以采用多一次 Map/Reduce 的方法， 避免倾斜。

最后得出的结论是：用 job 数的增加，输入量的增加，占用更多存储空间，充分利用空闲 CPU 等各种方法，分解数据倾斜造成的负担。

### 2、如何优化 ###

Hive系统内部已针对不同的查询预设定了优化方法，用户可以通过调整配置进行控制， 以下举例介绍部分优化的策略以及优化控制选项。

*  列裁剪

Hive 在读数据的时候，可以只读取查询中所需要用到的列，而忽略其它列。 例如，若有以下查询：

SELECT a,b FROM q WHERE e<10;

在实施此项查询中，Q 表有 5 列（a，b，c，d，e），Hive 只读取查询逻辑中真实需要 的 3 列 a、b，e而忽略列 c，d；这样做节省了读取开销，中间表存储开销和数据整合开销。  
裁剪所对应的参数项为：`hive.optimize.cp=true`（默认值为真）

*  分区裁剪

可以在查询的过程中减少不必要的分区。 例如，若有以下查询：

SELECT * FROM (SELECTT a1,COUNT(1) FROM T GROUP BY a1) subq WHERE subq.prtn=100; #（多余分区）

SELECT * FROM T1 JOIN (SELECT * FROM T2) subq ON (T1.a1=subq.a2) WHERE subq.prtn=100;

查询语句若将“subq.prtn=100”条件放入子查询中更为高效，可以减少读入的分区 数目。 Hive 自动执行这种裁剪优化。  
分区参数为：`hive.optimize.pruner=true`（默认值为真）

*  JOIN原则

在使用写有 Join 操作的查询语句时有一条原则：应该将条目少的表/子查询放在 Join 操作符的左边。原因是在 Join 操作的 Reduce 阶段，位于 Join 操作符左边的表的内容会被加载进内存，将条目少的表放在左边，可以有效减少发生 OOM 错误的几率。对于一条语句中有多个 Join 的情况，如果 Join 的条件相同，比如查询：

INSERT OVERWRITE TABLE pv_users 
     SELECT pv.pageid, u.age FROM page_view p 
     JOIN user u ON (pv.userid = u.userid) 
     JOIN newuser x ON (u.userid = x.userid);

如果 Join 的 key 相同，不管有多少个表，都会则会合并为一个 Map-Reduce。  
如果 Join 的条件不相同，比如：

INSERT OVERWRITE TABLE pv_users 
       SELECT pv.pageid, u.age FROM page_view p 
       JOIN user u ON (pv.userid = u.userid) 
       JOIN newuser x on (u.age = x.age);

Map-Reduce 的任务数目和 Join 操作的数目是对应的，上述查询和以下查询是等价的：

INSERT OVERWRITE TABLE tmptable 
       SELECT * FROM page_view p JOIN user u 
       ON (pv.userid = u.userid);

INSERT OVERWRITE TABLE pv_users 
       SELECT x.pageid, x.age FROM tmptable x 
       JOIN newuser y ON (x.age = y.age);

*  MAP JOIN操作

Join 操作在 Map 阶段完成，不再需要Reduce，前提条件是需要的数据在 Map 的过程中可以访问到。比如查询：

INSERT OVERWRITE TABLE pv_users 
       SELECT /*+ MAPJOIN(pv) */ pv.pageid, u.age 
       FROM page_view pv 
         JOIN user u ON (pv.userid = u.userid);

可以在 Map 阶段完成 Join，如图所示：  
![这里写图片描述][20170619092621360]  
相关的参数为：

hive.join.emit.interval = 1000 
    hive.mapjoin.size.key = 10000
    hive.mapjoin.cache.numrows = 10000

*  GROUP BY操作

进行GROUP BY操作时需要注意一下几点：  
（1）Map端部分聚合  
事实上并不是所有的聚合操作都需要在reduce部分进行，很多聚合操作都可以先在Map端进行部分聚合，然后reduce端得出最终结果。  
这里需要修改的参数为：`hive.map.aggr=true`（用于设定是否在 map 端进行聚合，默认值为真） `hive.groupby.mapaggr.checkinterval=100000`（用于设定 map 端进行聚合操作的条目数）  
（2）有数据倾斜时进行负载均衡  
此处需要设定 `hive.groupby.skewindata`，当选项设定为 true 是，生成的查询计划有两 个 MapReduce 任务。在第一个 MapReduce 中，map 的输出结果集合会随机分布到 reduce 中， 每个 reduce 做部分聚合操作，并输出结果。这样处理的结果是，相同的 Group By Key 有可 能分发到不同的 reduce 中，从而达到负载均衡的目的；第二个 MapReduce 任务再根据预处 理的数据结果按照 Group By Key 分布到 reduce 中（这个过程可以保证相同的 Group By Key 分布到同一个 reduce 中），最后完成最终的聚合操作。

*  合并小文件

我们知道文件数目小，容易在文件存储端造成瓶颈，给 HDFS 带来压力，影响处理效率。对此，可以通过合并Map和Reduce的结果文件来消除这样的影响。  
用于设置合并属性的参数有：  
是否合并Map输出文件：`hive.merge.mapfiles=true`（默认值为真）  
是否合并Reduce 端输出文件：`hive.merge.mapredfiles=false`（默认值为假）  
合并文件的大小：`hive.merge.size.per.task=256*1000*1000`（默认值为 256000000）

### 3、示例 ###

*  使用SQL提高查询

熟练地使用 SQL，能写出高效率的查询语句。  
场景：有一张 user 表，为卖家每天收到表，user\_id，ds（日期）为 key，属性有主营类目，指标有交易金额，交易笔数。每天要取前10天的总收入，总笔数，和最近一天的主营类目。 　　  
解决方法 1，如下所示，一般方法：

INSERT OVERWRITE TABLE t1 
    SELECT user_id,substr(MAX(CONCAT(ds,cat),9) AS main_cat) FROM users 
    WHERE ds=20120329 // 20120329 为日期列的值，实际代码中可以用函数表示出当天日期 GROUP BY user_id; 
    
    INSERT OVERWRITE TABLE t2 
    SELECT user_id,sum(qty) AS qty,SUM(amt) AS amt FROM users 
    WHERE ds BETWEEN 20120301 AND 20120329 
    GROUP BY user_id 
    
    SELECT t1.user_id,t1.main_cat,t2.qty,t2.amt FROM t1 
    JOIN t2 ON t1.user_id=t2.user_id

下面给出方法1的思路，实现步骤如下：  
第一步：利用分析函数，取每个 user\_id 最近一天的主营类目，存入临时表 t1。  
第二步：汇总 10 天的总交易金额，交易笔数，存入临时表 t2。  
第三步：关联 t1，t2，得到最终的结果。

解决方法 2，如下所示：优化方法

SELECT user_id,substr(MAX(CONCAT(ds,cat)),9) AS main_cat,SUM(qty),SUM(amt) FROM users 
    WHERE ds BETWEEN 20120301 AND 20120329 
    GROUP BY user_id

实际测试中发现，方案 2 的开销等于方案 1 的第二步的开销，性能提升，由原有的 25 分钟完成，缩短为 10 分钟以内完成。节省了两个临时表的读写是一个关键原因，这种方式也适用于 Oracle 中的数据查找工作。 SQL 具有普适性，很多 SQL 通用的优化方案在 Hadoop 分布式计算方式中也可以达到效果。

*  无效ID在关联时的数据倾斜问题

问题：日志中常会出现信息丢失，比如每日约为 20 亿的全网日志，其中的 user\_id 为主 键，在日志收集过程中会丢失，出现主键为 null 的情况，如果取其中的 user\_id 和 bmw\_users 关联，就会碰到数据倾斜的问题。原因是 Hive 中，主键为 null 值的项会被当做相同的 Key 而分配进同一个计算 Map。  
解决方法 1：user\_id 为空的不参与关联，子查询过滤 null

SELECT * FROM log a 
    JOIN bmw_users b ON a.user_id IS NOT NULL AND a.user_id=b.user_id 
    UNION All SELECT * FROM log a WHERE a.user_id IS NULL

解决方法 2 如下所示：函数过滤 null

SELECT * FROM log a LEFT OUTER 
    JOIN bmw_users b ON 
    CASE WHEN a.user_id IS NULL THEN CONCAT(‘dp_hive’,RAND()) ELSE a.user_id END =b.user_id;

调优结果：原先由于数据倾斜导致运行时长超过 1 小时，解决方法 1 运行每日平均时长 25 分钟，解决方法 2 运行的每日平均时长在 20 分钟左右。优化效果很明显。  
解决方法2比解决方法1效果更好，不但IO少了，而且作业数也少了。解决方法1中log读取两次，job 数为2。解决方法2中 job 数是1。这个优化适合无效 id（比如-99、 ‘’，null 等）产生的倾斜问题。把空值的 key 变成一个字符串加上随机数，就能把倾斜的 数据分到不同的Reduce上，从而解决数据倾斜问题。因为空值不参与关联，即使分到不同 的 Reduce 上，也不会影响最终的结果。附上 Hadoop 通用关联的实现方法是：关联通过二次排序实现的，关联的列为 partion key，关联的列和表的 tag 组成排序的 group key，根据 pariton key分配Reduce。同一Reduce内根据group key排序。

*  不同数据类型关联产生的倾斜问题

问题：不同数据类型 id 的关联会产生数据倾斜问题。  
一张表 s8 的日志，每个商品一条记录，要和商品表关联。但关联却碰到倾斜的问题。 s8 的日志中有 32 位字符串商品 id，也有数值商品 id，日志中类型是 string 的，但商品中的 数值 id 是 bigint 的。猜想问题的原因是把 s8 的商品 id 转成数值 id 做 hash 来分配 Reduce， 所以字符串 id 的 s8 日志，都到一个 Reduce 上了，解决的方法验证了这个猜测。  
解决方法：把数据类型转换成字符串类型

SELECT * FROM s8_log a LEFT OUTER 
    JOIN r_auction_auctions b ON a.auction_id=CASE(b.auction_id AS STRING)

调优结果显示：数据表处理由 1 小时 30 分钟经代码调整后可以在 20 分钟内完成。

*  利用Hive对UNION ALL优化的特性

多表 union all 会优化成一个 job。  
问题：比如推广效果表要和商品表关联，效果表中的 auction\_id 列既有 32 为字符串商 品 id，也有数字 id，和商品表关联得到商品的信息。  
解决方法：Hive SQL 性能会比较好

SELECT * FROM effect a 
    JOIN 
    (SELECT auction_id AS auction_id FROM auctions 
    UNION All 
    SELECT auction_string_id AS auction_id FROM auctions) b 
    ON a.auction_id=b.auction_id

比分别过滤数字 id，字符串 id 然后分别和商品表关联性能要好。  
这样写的好处：1 个 MapReduce 作业，商品表只读一次，推广效果表只读取一次。把 这个 SQL 换成 Map/Reduce 代码的话，Map 的时候，把 a 表的记录打上标签 a，商品表记录 每读取一条，打上标签 b，变成两个

SELECT * FROM 
    (SELECT * FROM t1 GROUP BY c1,c2,c3 UNION ALL SELECT * FROM t2 GROUP BY c1,c2,c3)t3 
    GROUP BY c1,c2,c3

从业务逻辑上说，子查询内的 GROUP BY 怎么都看显得多余（功能上的多余，除非有 COUNT(DISTINCT)），如果不是因为 Hive Bug 或者性能上的考量（曾经出现如果不执行子查询 GROUP BY，数据得不到正确的结果的 Hive Bug）。所以这个 Hive 按经验转换成如下所示：

SELECT * FROM (SELECT * FROM t1 UNION ALL SELECT * FROM t2)t3 GROUP BY c1,c2,c3

调优结果：经过测试，并未出现 union all 的 Hive Bug，数据是一致的。MapReduce 的 作业数由 3 减少到 1。 t1 相当于一个目录，t2 相当于一个目录，对 Map/Reduce 程序来说，t1，t2 可以作为 Map/Reduce 作业的 mutli inputs。这可以通过一个 Map/Reduce 来解决这个问题。Hadoop 的 计算框架，不怕数据多，就怕作业数多。  
但如果换成是其他计算平台如 Oracle，那就不一定了，因为把大的输入拆成两个输入， 分别排序汇总后 merge（假如两个子排序是并行的话），是有可能性能更优的（比如希尔排 序比冒泡排序的性能更优）。  
消灭子查询内的 COUNT(DISTINCT)，MAX，MIN。

SELECT * FROM 
    (SELECT * FROM t1 
    UNION ALL SELECT c1,c2,c3 COUNT(DISTINCT c4) FROM t2 GROUP BY c1,c2,c3) t3 
    GROUP BY c1,c2,c3;

由于子查询里头有 COUNT(DISTINCT)操作，直接去 GROUP BY 将达不到业务目标。这时采用 临时表消灭 COUNT(DISTINCT)作业不但能解决倾斜问题，还能有效减少 jobs。

INSERT t4 SELECT c1,c2,c3,c4 FROM t2 GROUP BY c1,c2,c3; 
    SELECT c1,c2,c3,SUM(income),SUM(uv) FROM 
    (SELECT c1,c2,c3,income,0 AS uv FROM t1 
    UNION ALL 
    SELECT c1,c2,c3,0 AS income,1 AS uv FROM t2) t3 
    GROUP BY c1,c2,c3;

job 数是 2，减少一半，而且两次 Map/Reduce 比 COUNT(DISTINCT)效率更高。  
调优结果：千万级别的类目表，member 表，与 10 亿级得商品表关联。原先 1963s 的任务经过调整，1152s 即完成。  
消灭子查询内的 JOIN

SELECT * FROM 
    (SELECT * FROM t1 UNION ALL SELECT * FROM t4 UNION ALL SELECT * FROM t2 JOIN t3 ON t2.id=t3.id) x 
    GROUP BY c1,c2;

上面代码运行会有 5 个 jobs。加入先 JOIN 生存临时表的话 t5，然后 UNION ALL，会变成 2 个 jobs。

INSERT OVERWRITE TABLE t5 
    SELECT * FROM t2 JOIN t3 ON t2.id=t3.id; 
    SELECT * FROM (t1 UNION ALL t4 UNION ALL t5);

调优结果显示：针对千万级别的广告位表，由原先 5 个 Job 共 15 分钟，分解为 2 个 job 一个 8-10 分钟，一个3分钟。

*  GROUP BY替代COUNT(DISTINCT)达到优化效果

计算 uv 的时候，经常会用到 COUNT(DISTINCT)，但在数据比较倾斜的时候 COUNT(DISTINCT) 会比较慢。这时可以尝试用 GROUP BY 改写代码计算 uv。  
原有代码

INSERT OVERWRITE TABLE s_dw_tanx_adzone_uv PARTITION (ds=20120329) 
    SELECT 20120329 AS thedate,adzoneid,COUNT(DISTINCT acookie) AS uv FROM s_ods_log_tanx_pv t WHERE t.ds=20120329 GROUP BY adzoneid

关于COUNT(DISTINCT)的数据倾斜问题不能一概而论，要依情况而定，下面是我测试的一组数据：  
测试数据：169857条

#统计每日IP 
    CREATE TABLE ip_2014_12_29 AS SELECT COUNT(DISTINCT ip) AS IP FROM logdfs WHERE logdate='2014_12_29';

耗时：24.805 seconds

#统计每日IP（改造） 
    CREATE TABLE ip_2014_12_29 AS SELECT COUNT(1) AS IP FROM (SELECT DISTINCT ip from logdfs WHERE logdate='2014_12_29') tmp;

耗时：46.833 seconds  
测试结果表明：明显改造后的语句比之前耗时，这是因为改造后的语句有2个SELECT，多了一个job，这样在数据量小的时候，数据不会存在倾斜问题。

[20170618160030273]: /images/20220614/9328104c8ecc441d97e21d805dc66b48.png
[20170618160655066]: /images/20220614/eb649f84e76641d8979c7ae5dc03287e.png
[20170618162221737]: /images/20220614/81e67961781a4c618edabc91e650a66c.png
[20170618165138268]: /images/20220614/82ed6a593ac6499ea95624ec0f0db861.png
[20170618163917543]: /images/20220614/3720583a4c8644b7b8da402947b00b9b.png
[20170618164023263]: /images/20220614/eeb7c0a9dc554a1a920d1ee13347e57e.png
[20170618165254581]: /images/20220614/8175bc8c4c8d4961844afe622ecc341e.png
[20170618165540817]: /images/20220614/8e410fd06ab8435fadaff986e986f121.png
[20170618165552395]: /images/20220614/448887bd92f84d07bf352f8ffc8ad7b9.png
[20170618165734958]: /images/20220614/63d430a5b51f452ca9750721adebc87a.png
[20170618165825959]: /images/20220614/988f571637f241b3a75108b4f1ea4029.png
[20170618171443861]: /images/20220614/349aef9572584e6685658f96a7b45a0b.png
[20170618171608784]: /images/20220614/fc5459a6ee3b4d828221757cf5045c19.png
[20170618213335820]: /images/20220614/9690096ac491422da0c098aaa5bff0f6.png
[20170618213622835]: /images/20220614/6e2909102861452f84bf93a91a9d04c8.png
[20170618213929172]: /images/20220614/43f4825f50fb4bb7a4f071e82e84ecb2.png
[20170618214333225]: /images/20220614/c1b734390327457abe1e90d22b76645a.png
[20170619092621360]: /images/20220614/517e7de2605142a39b039d4af816924a.png