hive

hive 内部表和外部表的区别

未被 external 修饰的是内部表（managed table）
被 external 修饰的为外部表（external table）

CREATE TABLE IF NOT EXISTS test_internal_table(logs array<struct<name:string, rpid:string, bid:string, uid:string, did:string, duid:string, hbuid:string, ua:string, device_id:string, ip:string, server_timestamp:BIGINT>>, level STRING, message STRING, client_timestamp BIGINT)
partitioned by (dt string)
ROW FORMAT SERDE 'org.openx.data.jsonserde.JsonSerDe'
STORED AS TEXTFILE;

CREATE EXTERNAL TABLE IF NOT EXISTS test_external_table(logs array<struct<name:string, rpid:string, bid:string, uid:string, did:string, duid:string, hbuid:string, ua:string, device_id:string, ip:string, server_timestamp:BIGINT>>, level STRING, message STRING, client_timestamp BIGINT)
partitioned by (dt string)
ROW FORMAT SERDE 'org.openx.data.jsonserde.JsonSerDe'
STORED AS TEXTFILE
LOCATION '/flume/events/birdben.ad.view_ad';

区别：

1. 建表时带有external关键字为外部表，否则为内部表内部表数据由 Hive 自身管理，外部表数据由 HDFS 管理；
2. 内部表和外部表建表时都可以自己指定location；内部表数据存储的位置是 hive.metastore.warehouse.dir（默认：/user/hive/warehouse），外部表数据的存储位置由自己制定（如果没有 LOCATION，Hive 将在 HDFS 上的 / user/hive/warehouse 文件夹下以外部表的表名创建一个文件夹，并将属于这个表的数据存放在这里）；
3. 删除内部表会直接删除元数据（metadata）及存储数据；删除外部表仅仅会删除元数据，HDFS 上的文件并不会被删除；

hive的metastore的三种模式

• 内嵌Derby方式

  这个是Hive默认的启动模式，一般用于单元测试，这种存储方式有一个缺点：在同一时间只能有一个进程连接使用数据库。

• Local方式

  本地MySQL

• Remote方式

  远程MySQL,一般常用此种方式

hive四种排序方式的区别

• order by
  order by 是要对输出的结果进行全局排序，这就意味着只有一个reducer才能实现（多个reducer无法保证全局有序）但是当数据量过大的时候，效率就很低。如果在严格模式下（hive.mapred.mode=strict）,则必须配合limit使用

• sort by
  sort by 不是全局排序，只是在进入到reducer之前完成排序，只保证了每个reducer中数据按照指定字段的有序性，是局部排序。配置mapred.reduce.tasks=[nums]可以对输出的数据执行归并排序。可以配合limit使用，提高性能

• distribute by
  distribute by 指的是按照指定的字段划分到不同的输出reduce文件中，和sort by一起使用时需要注意，distribute by必须放在前面

• cluster by
  cluster by 可以看做是一个特殊的distribute by+sort by，它具备二者的功能，但是只能实现倒序排序的方式,不能指定排序规则为asc 或者desc

Impala 和 hive 的查询有哪些区别

Impala是基于Hive的大数据实时分析查询引擎，直接使用Hive的元数据库Metadata,意味着impala元数据都存储在Hive的metastore中。并且impala兼容Hive的sql解析，实现了Hive的SQL语义的子集，功能还在不断的完善中。

Impala相对于Hive所使用的优化技术

• 1、没有使用 MapReduce进行并行计算，虽然MapReduce是非常好的并行计算框架，但它更多的面向批处理模式，而不是面向交互式的SQL执行。与 MapReduce相比：Impala把整个查询分成一执行计划树，而不是一连串的MapReduce任务，在分发执行计划后，Impala使用拉式获取数据的方式获取结果，把结果数据组成按执行树流式传递汇集，减少的了把中间结果写入磁盘的步骤，再从磁盘读取数据的开销。Impala使用服务的方式避免 每次执行查询都需要启动的开销，即相比Hive没了MapReduce启动时间。
• 2、使用LLVM产生运行代码，针对特定查询生成特定代码，同时使用Inline的方式减少函数调用的开销，加快执行效率。
• 3、充分利用可用的硬件指令（SSE4.2）。
• 4、更好的IO调度，Impala知道数据块所在的磁盘位置能够更好的利用多磁盘的优势，同时Impala支持直接数据块读取和本地代码计算checksum。
• 5、通过选择合适的数据存储格式可以得到最好的性能（Impala支持多种存储格式）。
• 6、最大使用内存，中间结果不写磁盘，及时通过网络以stream的方式传递。

Hive Sql 是怎样解析成MR job的?

主要分为6个阶段:

1. Hive使用Antlr实现语法解析.根据Antlr制定的SQL语法解析规则,完成SQL语句的词法/语法解析,将SQL转为 抽象语法树 AST.

2. 遍历AST,生成基本查询单元QueryBlock.QueryBlock是一条SQL最基本的组成单元，包括三个部分：输入源，计算过程，输出.

3. 遍历QueryBlock,生成OperatorTree.Hive最终生成的MapReduce任务，Map阶段和Reduce阶段均由OperatorTree组成。Operator就是在Map阶段或者Reduce阶段完成单一特定的操作。QueryBlock生成Operator Tree就是遍历上一个过程中生成的QB和QBParseInfo对象的保存语法的属性.

4. **优化OperatorTree.**大部分逻辑层优化器通过变换OperatorTree，合并操作符，达到减少MapReduce Job，减少shuffle数据量的目的

5. OperatorTree生成MapReduce Job.遍历OperatorTree,翻译成MR任务.

   • 对输出表生成MoveTask
   • 从OperatorTree的其中一个根节点向下深度优先遍历
   • ReduceSinkOperator标示Map/Reduce的界限，多个Job间的界限
   • 遍历其他根节点，遇过碰到JoinOperator合并MapReduceTask
   • 生成StatTask更新元数据
   • 剪断Map与Reduce间的Operator的关系

6. 优化任务. 使用物理优化器对MR任务进行优化,生成最终执行任务

hive 有索引吗

Hive 支持索引，但是 Hive 的索引与关系型数据库中的索引并不相同，比如，Hive 不支持主键或者外键。
Hive 索引可以建立在表中的某些列上，以提升一些操作的效率，例如减少 MapReduce 任务中需要读取的数据块的数量。在可以预见到分区数据非常庞大的情况下，索引常常是优于分区的。
虽然 Hive 并不像事物数据库那样针对个别的行来执行查询、更新、删除等操作。它更多的用在多任务节点的场景下，快速地全表扫描大规模数据。但是在某些场景下，建立索引还是可以提高 Hive 表指定列的查询速度。（虽然效果差强人意）

CREATE INDEX index_name 
ON TABLE base_table_name (col_name, ...)
AS 'index.handler.class.name'
[ WITH DEFERRED REBUILD]
[IDXPROPERTIES (property_name=property_value, ...)]
[ IN TABLE index_table_name]
[PARTITIONED BY (col_name, ...)]
[
   [ ROW FORMAT ...] STORED AS ...
   | STORED BY ...
]
[LOCATION hdfs_path]
[TBLPROPERTIES (...)]
[ COMMENT "index comment"]

• 索引适用的场景
  适用于不更新的静态字段。以免总是重建索引数据。每次建立、更新数据后，都要重建索引以构建索引表。
• Hive 索引的机制如下：
  hive 在指定列上建立索引，会产生一张索引表（Hive 的一张物理表），里面的字段包括，索引列的值、该值对应的 HDFS 文件路径、该值在文件中的偏移量;
  v0.8 后引入 bitmap 索引处理器，这个处理器适用于排重后，值较少的列（例如，某字段的取值只可能是几个枚举值）

因为索引是用空间换时间，索引列的取值过多会导致建立 bitmap 索引表过大。但是，很少遇到 hive 用索引的。说明还是有缺陷 or 不合适的地方的。

运维如何对 hive 进行调度

1. 将 hive 的 sql 定义在脚本当中
2. 使用 azkaban 或者 oozie 进行任务的调度
3. 监控任务调度页面

ORC、Parquet 等列式存储的优点

存储格式	存储方式	特点
TextFile	行存储	存储空间消耗比较大，并且压缩的 text 无法分割和合并 查询的效率最低, 可以直接存储，加载数据的速度最高
SequenceFile	行存储	存储空间消耗最大, 压缩的文件可以分割和合并 查询效率高，需要通过 text 文件转化来加载
RCFile	数据按行分块 每块按照列存储	存储空间最小，查询的效率最高 ，需要通过 text 文件转化来加载，加载的速度最低。压缩快 快速列存取。读记录尽量涉及到的 block 最少 读取需要的列只需要读取每个 row group 的头部定义。 读取全量数据的操作 性能可能比 sequencefile 没有明显的优势
ORCFile	数据按行分块 每块按照列存储	压缩快, 快速列存取 , 效率比 rcfile 高, 是 rcfile 的改良版本
Parquet	列存储	相对于 PRC，Parquet 压缩比较低，查询效率较低，不支持 update、insert 和 ACID. 但是 Parquet 支持 Impala 查询引擎

Hive支持ORCfile，这是一种新的表格存储格式，通过诸如谓词下推，压缩等技术来提高执行速度提升。对于每个HIVE表使用ORCFile应该是一件容易的事情，并且对于获得HIVE查询的快速响应时间非常有益。

	Parquethttp://parquet.apache.org	Orchttp://orc.apache.org
发展状态	目前都是 Apache 开源的顶级项目，列式存储引擎	目前都是 Apache 开源的顶级项目，列式存储引擎
开发语言	Java	Java
主导公司	Twitter/Cloudera	Hortonworks
ACID	不支持	支持 ACID 事务
修改操作 (update,delete)	不支持	支持
支持索引 (统计信息)	粗粒度索引block/group/chunk 级别统计信息	粗粒度索引file/stripe/row 级别统计信息，不能精确到列建立索引
支持的查询引擎	Apache Drill、Impala	Apache Hive
查询性能	Orc 性能更高一点	Orc 性能更高一点
压缩比	Orc 压缩比更高 (见下图)	Orc 压缩比更高 (见下图)
列编码	支持多种编码，字典，RLE，Delta 等	支持主流编码，与 Parquet 类似

总结：如果仅仅是在 HIve 中存储和查询，建议使用 ORC 格式，如果在 Hive 中存储，而使用 Impala 查询，建议使用 Parquet。

数据建模用的哪些模型？

1. 星型模型
   星形模式 (Star Schema) 是最常用的维度建模方式。星型模式是以事实表为中心，所有的维度表直接连接在事实表上，像星星一样。
   星形模式的维度建模由一个事实表和一组维表成，且具有以下特点：
   a. 维表只和事实表关联，维表之间没有关联；
   b. 每个维表主键为单列，且该主键放置在事实表中，作为两边连接的外键；
   c. 以事实表为核心，维表围绕核心呈星形分布；

2. 雪花模型
   雪花模式 (Snowflake Schema) 是对星形模式的扩展。雪花模式的维度表可以拥有其他维度表的，虽然这种模型相比星型更规范一些，但是由于这种模型不太容易理解，维护成本比较高，而且性能方面需要关联多层维表，性能也比星型模型要低。所以一般不是很常用。

3. 星座模型
   星座模式是星型模式延伸而来，星型模式是基于一张事实表的，而星座模式是基于多张事实表的，而且共享维度信息。前面介绍的两种维度建模方法都是多维表对应单事实表，但在很多时候维度空间内的事实表不止一个，而一个维表也可能被多个事实表用到。在业务发展后期，绝大部分维度建模都采用的是星座模式。

为什么要对数据仓库分层？

• 用空间换时间，通过大量的预处理来提升应用系统的用户体验（效率），因此数据仓库会存在大量冗余的数据。
• 如果不分层的话，如果源业务系统的业务规则发生变化将会影响整个数据清洗过程，分散工作量。
• 通过数据分层管理可以简化数据清洗的过程，因为把原来一步的工作分到了多个步骤去完成，相当于把一个复杂的工作拆成了多个简单的工作，把一个大的黑盒变成了一个白盒，每一层的处理逻辑都相对简单和容易理解，这样我们比较容易保证每一个步骤的正确性，当数据发生错误的时候，往往我们只需要局部调整某个步骤即可。

分层的主要原因是在管理数据的时候,能对数据有一个更加清晰的掌控,详细来讲,主要有下面几个原因:

• 清晰数据结构：每一个数据分层都有它的作用域,在使用表的时候能更方便地定位和理解。

• 数据血缘追踪：简单来说,我们最终给业务呈现的是-个能直接使用业务表,但是它的来源有很多,如果有- -张来源表出问题了 ,我们希望能够快速准确地定位到问题,并清楚它的危害范围。

• 减少重复开发：规范数据分层, 开发一些通用的中间层数据 ,能够减少极大的重复计算。

• 把复杂问题简单化：将一个复杂的任务分解成多个步骤来完成,每一-层只处理单-的步骤,比较简单和容易理解。而且便于维护数据的准确性，当数据出现问题之后,可以不用修复所有的数据,只需要从有问题的步骤开始修复。

• 屏蔽原始数据的异常：屏蔽业务的影响,不必改- -次业务就需要重新接入数据

• ODS:操作型数据(Operational Data Store)，指结构与源系统基本保持一致的增量或者全量数据。作为DW数据的一个数据准备区，同时又承担基础数据记录历史变化，之所以保留原始数据和线上原始数据保持一致，方便后期数据核对需要。

• CDM：通用数据模型，又称为数据中间层(Common Data Model)，包含DWD、DWS、DIM层。

  • DWD：数据仓库明细层数据(Data Warehouse Detail)。对ODS层数据进行清洗转化，以业务过程作为建模驱动，基于每个具体的业务过程特点，构建最细粒度的明细事实表。可以结合企业的数据使用特点，基于维度建模思想，将明细事实表的某些重要属性字段做适当冗余，也即宽表化处理，构建明细宽表。
  • DWS：数据仓库汇总层数据(Data Warehouse Summary)，基于指标需求，构建初步汇总事实表，一般是宽表。基于上层的应用和产品的指标需求，构建公共粒度的汇总指标表。以宽表化手段物理化模型，构建命名规范、口径一致的统计指标，为上层提供公共指标。
  • DIM：建立一致数据分析维表，可以降低数据计算口径不统一的风险，同时可以方便进行交叉探查。以维度作为建模驱动，基于每个维度的业务含义，通过添加维度属性、关联维度等定义计算逻辑，完成属性定义的过程并建立一致的数据分析维表。

• ADS：面向应用的数据服务层(Application Data Service)。整合汇总成分析某一个主题域的服务数据，面向应用逻辑的数据加工。该层主要存放数据产品个性化的统计指标数据，这一层的数据直接对接数据的消费者，是产品、运营等角色可以直接感知理解的一层，大多数这一层的表都可以直接在BI上通过图表的形式直接透出。

使用过 Hive 解析 JSON 串吗

• hive 处理 json 数据总体来说有两个方向的路走

1. 将 json 以字符串的方式整个入 Hive 表，然后通过使用 UDF函数解析已经导入到 hive 中的数据，比如使用 LATERAL VIEW json_tuple的方法，获取所需要的列名。

2. 在导入之前将 json 拆成各个字段，导入 Hive 表的数据是已经解析过得。这将需要使用第三方的 SerDe。（SerDe是Serialize/Deserilize的简称，目的是用于序列化和反序列化）

怎么排查是哪里出现了数据倾斜

原因

key分布不均匀
业务数据本身的特性
建表时考虑不周
某些SQL语句本身就有数据倾斜

关键词	情形	后果
Join	其中一个表较小，但是key集中	分发到某一个或几个reduce上的数据远高于平均值
Join	大表与大表，但是分桶的判断字段0值或空值过多	这些空值都由一个reduce处理，非常慢
group by	group by 维度过小，某值的数量过多	处理某值的reduce耗时
Count Distinct	某特殊值过多	处理此特殊值的reduce耗时

表现

任务进度长时间维持在99%（或100%），查看任务监控页面，发现只有少量（1个或几个）reduce子任务未完成。因为其处理的数据量和其他reduce差异过大；
单一reduce的记录数与平均记录数差异过大，通常可能达到3倍甚至更多；
最长时长远大于平均时长。

数据倾斜怎么解决、数据优化

参数调节

并行执行，调节 parallel 参数；
调节 jvm 参数，重用 jvm；
设置 map、reduce 的参数；开启 strict mode 模式；
关闭推测执行设置。

• set hive.map.aggr = true
  在map中会做部分聚集操作，效率更高但需要更多的内存。

• set hive.groupby.skewindata = true
  数据倾斜的时候进行负载均衡，查询计划生成两个MR job，第一个job先进行key随机分配处理，随机分布到Reduce中，每个Reduce做部分聚合操作，先缩小数据量。第二个job再进行真正的group by key处理，根据预处理的数据结果按照Group By Key分布到Reduce中（这个过程可以保证相同的Key被分布到同一个Reduce中），完成最终的聚合操作。

• set hive.merge.mapfiles=true
  当出现小文件过多，需要合并小文件

• set hive.exec.reducers.bytes.per.reducer=1000000000（单位是字节）
  每个reduce能够处理的数据量大小，默认是1G。

• hive.exec.reducers.max=999
  最大可以开启的reduce个数，默认是999个。在只配了hive.exec.reducers.bytes.per.reducer以及hive.exec.reducers.max的情况下，实际的reduce个数会根据实际的数据总量/每个reduce处理的数据量来决定。

• set mapred.reduce.tasks=-1
  实际运行的reduce个数，默认是-1，可以认为指定，但是如果认为在此指定了，那么就不会通过实际的总数据量hive.exec.reducers.bytes.per.reducer来决定reduce的个数了。

SQL语句优化

1.大小表Join
大表对小表：设置自动识别小表，将小表放入内存中去执行。

在小表和大表进行join时，将小表放在前边，效率会高，hive会将小表进行缓存
使用map join让小的维度表（1000条以下的记录条数） 先进内存，在map端完成reduce。如下：

select /*+ mapjoin(a) */ 
a.c1, b.c1 ,b.c2
from a join b 
where a.c1 = b.c1;

2.大表Join大表

大表对大表：尽量减少数据集，可以通过分区表，避免扫描全表或者全字段；

把空值的key变成一个字符串加上随机数，把倾斜的数据分到不同的reduce上，由于null值关联不上，处理后并不影响最终结果。如下：

select * from log a 
left outer join users b 
on 
case when a.user_id is null 
then concat('hive',rand()) 
else a.user_id end = b.user_id;

3.count distinct大量相同特殊值
count distinct时，将值为null的情况单独处理，如果是计算count distinct，可以不用处理，直接过滤，在最后结果中加1。如果还有其他计算，需要进行group by，可以先将值为空的记录单独处理，再和其他计算结果进行union。

执行如

select a,count(distinct b) from t group by a;

类型的SQL时，会出现数据倾斜的问题
可替换成

select a,sum(1) from (select a, b from t group by a,b) group by a;

4.group by维度过小
采用sum() group by的方式来替换count(distinct)完成计算。

5.不同数据类型关联产生数据倾斜
用户表中user_id字段为int，**log表中user_id字段既有string类型也有int类型。**当按照user_id进行两个表的Join操作时，默认的Hash操作会按int型的id来进行分配，这样会导致所有string类型id的记录都分配到一个Reducer中。

select * from users a
left outer join logs b
on a.usr_id = cast(b.user_id as string)

6.小表不小不大，怎么用 map join 解决倾斜问题
使用 map join 解决小表(记录数少)关联大表的数据倾斜问题，这个方法使用的频率非常高，但如果小表很大，大到map join会出现bug或异常，这时就需要特别的处理。 以下例子:

select * from log a
left outer join users b
on a.user_id = b.user_id;

users 表有 600w+ 的记录，把 users 分发到所有的 map 上也是个不小的开销，而且 map join 不支持这么大的小表。如果用普通的 join，又会碰到数据倾斜的问题。
解决方法：

select /*+mapjoin(x)*/* from log a
  left outer join (
    select  /*+mapjoin(c)*/d.*
      from ( select distinct user_id from log ) c
      join users d
      on c.user_id = d.user_id
    ) x
  on a.user_id = b.user_id;

数据格式：数据存储及压缩。

针对 hive 中表的存储格式通常有 orc 和 parquet，压缩格式一般使用 snappy。相比与 textfile 格式表，orc 占有更少的存储。因为 hive 底层使用 MR 计算架构，数据流是 hdfs 到磁盘再到 hdfs，而且会有很多次，所以使用 orc 数据格式和 snappy 压缩策略可以降低 IO 读写，还能降低网络传输量，这样在一定程度上可以节省存储，还能提升 hql 任务执行效率；

有效地减小数据集将大表拆分成子表；结合使用外部表和分区表。

使用过 Hive 解析 JSON 串吗

• hive 处理 json 数据总体来说有两个方向的路走

1. 将 json 以字符串的方式整个入 Hive 表，然后通过使用 UDF 函数解析已经导入到 hive 中的数据，比如使用 LATERAL VIEW json_tuple 的方法，获取所需要的列名。
2. 在导入之前将 json 拆成各个字段，导入 Hive 表的数据是已经解析过得。这将需要使用第三方的
   SerDe。

运维如何对 hive 进行调度

1. 将 hive 的 sql 定义在脚本当中
2. 使用 azkaban 或者 oozie 进行任务的调度
3. 监控任务调度页面

hive 小文件过多怎么解决

小文件是如何产生的

1. 动态分区插入数据，产生大量的小文件，从而导致map数量剧增。
2. reduce数量越多，小文件也越多(reduce的个数和输出文件是对应的)。
3. 数据源本身就包含大量的小文件。

小文件问题的影响

1. 从Hive的角度看，小文件会开很多map，一个map开一个JVM去执行，所以这些任务的初始化，启动，执行会浪费大量的资源，严重影响性能。
2. 在HDFS中，每个小文件对象约占150byte，如果小文件过多会占用大量内存。这样NameNode内存容量严重制约了集群的扩展。

小文件问题的解决方案

A、从小文件产生的途经就可以从源头上控制小文件数量，方法如下：

1. 使用Sequencefile作为表存储格式，不要用textfile，在一定程度上可以减少小文件。
2. 减少reduce的数量(可以使用参数进行控制)。
3. 少用动态分区，用时记得按distribute by分区。

B、对于已有的小文件，我们可以通过以下几种方案解决

1. 使用hadoop archive命令把小文件进行归档。
2. 重建表，建表时减少reduce数量。
3. 通过参数进行调节，设置map/reduce端的相关参数，如下：

C、设置map输入合并小文件的相关参数：

view plain copy
//每个Map最大输入大小(这个值决定了合并后文件的数量)  
set mapred.max.split.size=256000000;    
//一个节点上split的至少的大小(这个值决定了多个DataNode上的文件是否需要合并)  
set mapred.min.split.size.per.node=100000000;  
//一个交换机下split的至少的大小(这个值决定了多个交换机上的文件是否需要合并)    
set mapred.min.split.size.per.rack=100000000;  
//执行Map前进行小文件合并  
set hive.input.format=org.apache.hadoop.hive.ql.io.CombineHiveInputFormat;   

设置map输出和reduce输出进行合并的相关参数：
[java] view plain copy
//设置map端输出进行合并，默认为true  
set hive.merge.mapfiles = true  
//设置reduce端输出进行合并，默认为false  
set hive.merge.mapredfiles = true  
//设置合并文件的大小  
set hive.merge.size.per.task = 256*1000*1000  
//当输出文件的平均大小小于该值时，启动一个独立的MapReduce任务进行文件merge。  
set hive.merge.smallfiles.avgsize=16000000

D、Write good SQL

E、存储格式

可以使用列裁剪，分区裁剪，orc，parquet等存储格式。

作为一个例子，考虑两个大表A和B（作为文本文件存储，其中一些列未在此处指定，即行试存储的缺点）以及一个简单的查询，如：

SELECT A.customerID, A.name, A.age, A.address join

B.role, B.department, B.salary

ON A.customerID=B.customerID;

此查询可能需要很长时间才能执行，因为表A和B都以TEXT形式存储，进行全表扫描。将这些表格转换为ORCFile格式通常会显着减少查询时间：ORC支持压缩存储（使用ZLIB或如上所示使用SNAPPY），但也支持未压缩的存储。

CREATE TABLE A_ORC (
customerID int, name string, age int, address string
) STORED AS ORC tblproperties (“orc.compress" = “SNAPPY”);
INSERT INTO TABLE A_ORC SELECT * FROM A;
CREATE TABLE B_ORC (
customerID int, role string, salary float, department string
) STORED AS ORC tblproperties (“orc.compress" = “SNAPPY”);

INSERT INTO TABLE B_ORC SELECT * FROM B;

SELECT A_ORC.customerID, A_ORC.name,
A_ORC.age, A_ORC.address join
B_ORC.role, B_ORC.department, B_ORC.salary
ON A_ORC.customerID=B_ORC.customerID;

F、压缩格式

大数据场景下存储格式压缩格式尤为关键，可以提升计算速度，减少存储空间，降低网络io，磁盘io，所以要选择合适的压缩格式和存储格式

压缩格式	UNIX 工具	算 法	文件扩展名	可分割
DEFLATE	无	DEFLATE	.deflate	No
gzip	gzip	DEFLATE	.gz	No
LZ4	无	LZ4	.LZ4	NO
bzip	bzip	bzip	.bz2	YES
LZO	lzop	LZO	.lzo	YES if indexed
Snappy	无	Snappy	.snappy	NO

• gzip:
  优点：压缩比在四种压缩方式中较高；hadoop 本身支持，在应用中处理 gzip 格式的文件就和直接处理文本一样；有 hadoop native 库；大部分 linux 系统都自带 gzip 命令，使用方便。
  缺点：不支持 split。
• lzo 压缩
  优点：压缩 / 解压速度也比较快，合理的压缩率；支持 split，是 hadoop 中最流行的压缩格式；支持 hadoop native 库；需要在 linux 系统下自行安装 lzop 命令，使用方便。
  缺点：压缩率比 gzip 要低；hadoop 本身不支持，需要安装；lzo 虽然支持 split，但需要对 lzo 文件建索引，否则 hadoop 也是会把 lzo 文件看成一个普通文件（为了支持 split 需要建索引，需要指定 inputformat 为 lzo 格式）。
• snappy 压缩
  优点：压缩速度快；支持 hadoop native 库。
  缺点：不支持 split；压缩比低；hadoop 本身不支持，需要安装；linux 系统下没有对应的命令。
• bzip2 压缩
  优点：支持 split；具有很高的压缩率，比 gzip 压缩率都高；hadoop 本身支持，但不支持 native；在 linux 系统下自带 bzip2 命令，使用方便。
  缺点：压缩 / 解压速度慢；不支持 native。

Hive SQL : 按照学生科目取每个科目的TopN

id,name,subject,score
1,小明,语文,87
2,张三,语文,27
3,王五,语文,69
4,李四,语文,99
5,小明,数学,86
6,马六,数学,33
7,李四,数学,44
8,小红,数学,50

按照各个科目的成绩排名 取 Top3

select a.* from
(select id,name,subject,score,row_number() over(partition by subject order by score desc) rank from student) a
where a.rank <= 3

Hive SQL: 获取每个用户的前1/4次的数据

cookieId  createTime    pv
--------------------------
cookie1 2015-04-10      1
cookie1 2015-04-11      5
cookie1 2015-04-12      7
cookie1 2015-04-13      3
cookie1 2015-04-14      2
cookie1 2015-04-15      4
cookie1 2015-04-16      4
cookie2 2015-04-10      2
cookie2 2015-04-11      3
cookie2 2015-04-12      5
cookie2 2015-04-13      6
cookie2 2015-04-14      3
cookie2 2015-04-15      9
cookie2 2015-04-16      7

获取每个用户前1/4次的访问记录

SELECT a.* from 
(SELECT cookieid,createtime,pv,NTILE(4)
OVER(PARTITION BY cookieId ORDER BY createtime) AS rn
from table ) a
WHERE a.rn = 1

NTILE(n)，用于将分组数据按照顺序切分成n片，返回当前切片值

Hive UDF简单介绍

在Hive中，用户可以自定义一些函数，用于扩展HiveQL的功能，而这类函数叫做UDF（用户自定义函数）。UDF分为两大类：UDAF（用户自定义聚合函数）和UDTF（用户自定义表生成函数）。

Hive有两个不同的接口编写UDF程序。一个是基础的UDF接口，一个是复杂的GenericUDF接口。

1. org.apache.hadoop.hive.ql. exec.UDF 基础UDF的函数读取和返回基本类型，即Hadoop和Hive的基本类型。如，Text、IntWritable、LongWritable、DoubleWritable等。
2. org.apache.hadoop.hive.ql.udf.generic.GenericUDF 复杂的GenericUDF可以处理Map、List、Set类型。

top K 问题

    前两天面试 3 面学长问我的这个问题（想说 TEG 的 3 个面试学长都是好和蔼，希望能完成最后一面，各方面原因造成我无比想去鹅场的心已经按捺不住了），这个问题还是建立最小堆比较好一些。

    先拿 10000 个数建堆，然后一次添加剩余元素，如果大于堆顶的数（10000 中最小的），将这个数替换堆顶，并调整结构使之仍然是一个最小堆，这样，遍历完后，堆中的 10000 个数就是所需的最大的 10000 个。建堆[时间复杂度](https://so.csdn.net/so/search?q=%E6%97%B6%E9%97%B4%E5%A4%8D%E6%9D%82%E5%BA%A6)是 O（mlogm），算法的时间复杂度为 O（nmlogm）（n 为 10 亿，m 为 10000）。

    优化的方法：可以把所有 10 亿个数据分组存放，比如分别放在 1000 个文件中。这样处理就可以分别在每个文件的 10^6 个数据中找出最大的 10000 个数，合并到一起在再找出最终的结果。

    以上就是面试时简单提到的内容，下面整理一下这方面的问题：



    在大规模数据处理中，经常会遇到的一类问题：在海量数据中找出出现频率最好的前 k 个数，或者从海量数据中找出最大的前 k 个数，这类问题通常被称为 top K 问题。例如，在搜索引擎中，统计搜索最热门的 10 个查询词；在歌曲库中统计下载最高的前 10 首歌等。

    针对 top K 类问题，通常比较好的方案是分治 + Trie 树 / hash + 小顶堆（就是上面提到的最小堆），即先将数据集按照 Hash 方法分解成多个小数据集，然后使用 Trie 树活着 Hash 统计每个小数据集中的 query 词频，之后用小顶堆求出每个数据集中出现频率最高的前 K 个数，最后在所有 top K 中求出最终的 top K。

eg：有 1 亿个浮点数，如果找出期中最大的 10000 个？

    最容易想到的方法是将数据全部排序，然后在排序后的集合中进行查找，最快的排序算法的时间复杂度一般为 O（nlogn），如快速排序。但是在 32 位的机器上，每个 float 类型占 4 个字节，1 亿个浮点数就要占用 400MB 的存储空间，对于一些可用内存小于 400M 的计算机而言，很显然是不能一次将全部数据读入内存进行排序的。其实即使内存能够满足要求（我机器内存都是 8GB），该方法也并不高效，因为题目的目的是寻找出最大的 10000 个数即可，而排序却是将所有的元素都排序了，做了很多的无用功。

    第二种方法为局部淘汰法，该方法与排序方法类似，用一个容器保存前 10000 个数，然后将剩余的所有数字——与容器内的最小数字相比，如果所有后续的元素都比容器内的 10000 个数还小，那么容器内这个 10000 个数就是最大 10000 个数。如果某一后续元素比容器内最小数字大，则删掉容器内最小元素，并将该元素插入容器，最后遍历完这 1 亿个数，得到的结果容器中保存的数即为最终结果了。此时的时间复杂度为 O（n+m^2），其中 m 为容器的大小，即 10000。

    第三种方法是分治法，将 1 亿个数据分成 100 份，每份 100 万个数据，找到每份数据中最大的 10000 个，最后在剩下的 100*10000 个数据里面找出最大的 10000 个。如果 100 万数据选择足够理想，那么可以过滤掉 1 亿数据里面 99% 的数据。100 万个数据里面查找最大的 10000 个数据的方法如下：用快速排序的方法，将数据分为 2 堆，如果大的那堆个数 N 大于 10000 个，继续对大堆快速排序一次分成 2 堆，如果大的那堆个数 N 大于 10000 个，继续对大堆快速排序一次分成 2 堆，如果大堆个数 N 小于 10000 个，就在小的那堆里面快速排序一次，找第 10000-n 大的数字；递归以上过程，就可以找到第 1w 大的数。参考上面的找出第 1w 大数字，就可以类似的方法找到前 10000 大数字了。此种方法需要每次的内存空间为 10^6*4=4MB，一共需要 101 次这样的比较。

    第四种方法是 Hash 法。如果这 1 亿个书里面有很多重复的数，先通过 Hash 法，把这 1 亿个数字去重复，这样如果重复率很高的话，会减少很大的内存用量，从而缩小运算空间，然后通过分治法或最小堆法查找最大的 10000 个数。

    第五种方法采用最小堆。首先读入前 10000 个数来创建大小为 10000 的最小堆，建堆的时间复杂度为 O（mlogm）（m 为数组的大小即为 10000），然后遍历后续的数字，并于堆顶（最小）数字进行比较。如果比最小的数小，则继续读取后续数字；如果比堆顶数字大，则替换堆顶元素并重新调整堆为最小堆。整个过程直至 1 亿个数全部遍历完为止。然后按照中序遍历的方式输出当前堆中的所有 10000 个数字。该算法的时间复杂度为 O（nmlogm），[空间复杂度](https://so.csdn.net/so/search?q=%E7%A9%BA%E9%97%B4%E5%A4%8D%E6%9D%82%E5%BA%A6)是 10000（常数）。

实际运行：

    实际上，最优的解决方案应该是最符合实际设计需求的方案，在时间应用中，可能有足够大的内存，那么直接将数据扔到内存中一次性处理即可，也可能机器有多个核，这样可以采用多线程处理整个数据集。

   下面针对不容的应用场景，分析了适合相应应用场景的解决方案。

（1）单机 + 单核 + 足够大内存

    如果需要查找 10 亿个查询次（每个占 8B）中出现频率最高的 10 个，考虑到每个查询词占 8B，则 10 亿个查询次所需的内存大约是 10^9 * 8B=8GB 内存。如果有这么大内存，直接在内存中对查询次进行排序，顺序遍历找出 10 个出现频率最大的即可。这种方法简单快速，使用。然后，也可以先用 HashMap 求出每个词出现的频率，然后求出频率最大的 10 个词。

（2）单机 + 多核 + 足够大内存

    这时可以直接在内存总使用 Hash 方法将数据划分成 n 个 partition，每个 partition 交给一个线程处理，线程的处理逻辑同（1）类似，最后一个线程将结果归并。

    该方法存在一个瓶颈会明显影响效率，即数据倾斜。每个线程的处理速度可能不同，快的线程需要等待慢的线程，最终的处理速度取决于慢的线程。而针对此问题，解决的方法是，将数据划分成 c×n 个 partition（c>1），每个线程处理完当前 partition 后主动取下一个 partition 继续处理，知道所有数据处理完毕，最后由一个线程进行归并。

（3）单机 + 单核 + 受限内存

    这种情况下，需要将原数据文件切割成一个一个小文件，如次啊用 hash(x)%M，将原文件中的数据切割成 M 小文件，如果小文件仍大于内存大小，继续采用 Hash 的方法对数据文件进行分割，知道每个小文件小于内存大小，这样每个文件可放到内存中处理。采用（1）的方法依次处理每个小文件。

（4）多机 + 受限内存

    这种情况，为了合理利用多台机器的资源，可将数据分发到多台机器上，每台机器采用（3）中的策略解决本地的数据。可采用 hash+socket 方法进行数据分发。

    从实际应用的角度考虑，（1）（2）（3）（4）方案并不可行，因为在大规模数据处理环境下，作业效率并不是首要考虑的问题，算法的扩展性和容错性才是首要考虑的。算法应该具有良好的扩展性，以便数据量进一步加大（随着业务的发展，数据量加大是必然的）时，在不修改算法框架的前提下，可达到近似的线性比；算法应该具有容错性，即当前某个文件处理失败后，能自动将其交给另外一个线程继续处理，而不是从头开始处理。  

    top K 问题很适合采用 MapReduce 框架解决，用户只需编写一个 Map 函数和两个 Reduce 函数，然后提交到 Hadoop（采用 Mapchain 和 Reducechain）上即可解决该问题。具体而言，就是首先根据数据值或者把数据 hash(MD5) 后的值按照范围划分到不同的机器上，最好可以让数据划分后一次读入内存，这样不同的机器负责处理不同的数值范围，实际上就是 Map。得到结果后，各个机器只需拿出各自出现次数最多的前 N 个数据，然后汇总，选出所有的数据中出现次数最多的前 N 个数据，这实际上就是 Reduce 过程。对于 Map 函数，采用 Hash 算法，将 Hash 值相同的数据交给同一个 Reduce task；对于第一个 Reduce 函数，采用 HashMap 统计出每个词出现的频率，对于第二个 Reduce 函数，统计所有 Reduce task，输出数据中的 top K 即可。  

    直接将数据均分到不同的机器上进行处理是无法得到正确的结果的。因为一个数据可能被均分到不同的机器上，而另一个则可能完全聚集到一个机器上，同时还可能存在具有相同数目的数据。  

以下是一些经常被提及的该类问题。

（1）有 10000000 个记录，这些查询串的重复度比较高，如果除去重复后，不超过 3000000 个。一个查询串的重复度越高，说明查询它的用户越多，也就是越热门。请统计最热门的 10 个查询串，要求使用的内存不能超过 1GB。

（2）有 10 个文件，每个文件 1GB，每个文件的每一行存放的都是用户的 query，每个文件的 query 都可能重复。按照 query 的频度排序。

（3）有一个 1GB 大小的文件，里面的每一行是一个词，词的大小不超过 16 个字节，内存限制大小是 1MB。返回频数最高的 100 个词。

（4）提取某日访问网站次数最多的那个 IP。

（5）10 亿个整数找出重复次数最多的 100 个整数。

（6）搜索的输入信息是一个字符串，统计 300 万条输入信息中最热门的前 10 条，每次输入的一个字符串为不超过 255B，内存使用只有 1GB。

（7）有 1000 万个身份证号以及他们对应的数据，身份证号可能重复，找出出现次数最多的身份证号。

重复问题

    在海量数据中查找出重复出现的元素或者去除重复出现的元素也是常考的问题。针对此类问题，一般可以通过位图法实现。例如，已知某个文件内包含一些电话号码，每个号码为 8 位数字，统计不同号码的个数。  

    本题最好的解决方法是通过使用位图法来实现。8 位整数可以表示的最大十进制数值为 99999999。如果每个数字对应于位图中一个 bit 位，那么存储 8 位整数大约需要 99MB。因为 1B=8bit，所以 99Mbit 折合成内存为 99/8=12.375MB 的内存，即可以只用 12.375MB 的内存表示所有的 8 位数电话号码的内容。

https://co-hwang.github.io/2019/07/15/Hive%E5%AD%A6%E4%B9%A0%E4%B9%8B%E8%B7%AF%EF%BC%88%E4%BA%8C%EF%BC%89Hive%E6%80%BB%E7%BB%93%E5%8F%8A%E4%BC%98%E5%8C%96/
https://monkeyip.github.io/2019/04/25/Hive%E6%95%B0%E6%8D%AE%E5%80%BE%E6%96%9C%E4%BC%98%E5%8C%96%E6%80%BB%E7%BB%93/