Kettle构建Hadoop ETL实践（四）：建立ETL示例模型

目录

一、业务场景

1. 操作型数据源

2. 销售订单数据仓库模型设计

二、HIVE相关配置

1. 选择文件格式

2. 选择表类型

3. 支持行级更新

4. Hive事务支持的限制

三、建立数据库表

1. 源数据库表

2. RDS库表

3. TDS库表

四、装载日期维度数据

五、小节

从本篇开始，介绍使用Kettle实现Hadoop数据仓库的ETL过程。我们会引入一个典型的订单业务场景作为示例，说明多维模型及其相关ETL技术在Kettle上的具体实现。本篇首先介绍一个小而典型的销售订单示例，描述业务场景，说明示例中包含的实体和关系，并在MySQL数据库上建立源数据库表并生成初始的数据。我们要在Hive中创建源数据过渡区和数据仓库的表，因此需要了解与Hive创建表相关的技术问题，包括使用Hive建立传统多维数据仓库时，如何选择适当的文件格式，Hive支持哪些表类型，向不同类型的表中装载数据时具有哪些不同特性。我们将以实验的方式对这些问题加以说明。在此基础上，我们就可以编写Hive的HiveQL脚本，建立过渡区和数据仓库中的表。本篇最后会说明日期维度的数据装载方式及其Kettle实现。

一、业务场景

1. 操作型数据源

示例的操作型系统是一个销售订单系统，初始时只有产品、客户、销售订单三个表，实体关系图如图4-1所示。

图4-1 销售订单源系统

这个场景中的表及其属性都很简单。产品表和客户表属于基本信息表，分别存储产品和客户的信息。产品只有产品编号、产品名称、产品分类三个属性，产品编号是主键，唯一标识一个产品。客户有六个属性，除客户编号和客户名称外，还包含省、市、街道、邮编四个客户所在地区属性。客户编号是主键，唯一标识一个客户。在实际应用中，基本信息表通常由其它后台系统维护。销售订单表有六个属性，订单号是主键，唯一标识一条销售订单记录。产品编号和客户编号是两个外键，分别引用产品表和客户表的主键。另外三个属性是订单时间、登记时间和订单金额。订单时间指的是客户下订单的时间，订单金额属性指的是该笔订单需要花费的金额，这些属性的含义很清楚。订单登记时间表示订单录入的时间，大多数情况下它应该等同于订单时间。如果由于某种情况需要重新录入订单，还要同时记录原始订单的时间和重新录入的时间，或者出现某种问题，订单登记时间滞后于下订单的时间，这两个属性值就会不同。本专题后面“迟到的事实”部分会讨论这种情况。

源系统采用关系模型设计，为了减少表的数量，这个系统只做到了2NF。地区信息依赖于邮编，所以这个模型中存在传递依赖。

2. 销售订单数据仓库模型设计

我们使用四步建模法设计星型数据仓库模型。 （1）选择业务流程。在本示例中只涉及一个销售订单的业务流程。 （2）声明粒度。ETL处理时间周期为每天一次，事实表中存储最细粒度的订单事务记录。 （3）确认维度。显然产品和客户是销售订单的维度。日期维度用于业务集成，并为数据仓库提供重要的历史视角，每个数据仓库中都应该有一个日期维度。订单维度是特意设计的，用于后面说明退化维度技术。我们将在后面详细介绍“退化维度”。 （4）确认事实。销售订单是当前场景中唯一的事实。

示例数据仓库的实体关系图如图4-2所示。

图4-2 销售订单数据仓库

作为演示示例，上面实体关系图中的实体属性都很简单，看属性名字便知其含义。除了日期维度外，其它三个维度都在源数据的基础上增加了代理键、版本号、生效日期、过期日期四个属性，用来描述维度变化的历史。当维度属性发生变化时，依据不同的策略，或生成一条新的维度记录，或直接修改原记录。日期维度有其特殊性，该维度数据一旦生成就不会改变，所以不需要版本号、生效日期和过期日期。代理键是维度表的主键。事实表引用维度表的代理键作为自己的外键，四个外键构成了事实表的联合主键。订单金额是当前事实表中的唯一度量。

二、HIVE相关配置

在“数据仓库架构中的ETL”曾经提到Hive可以用于原始数据和转换后的数据仓库数据存储。使用Hive作为多维数据仓库的主要挑战是处理渐变维（SCD）和生成代理键。处理渐变维需要配置Hive支持行级更新，并在建表时选择适当的文件格式。生成代理键在关系数据库中一般都是用自增列（如MySQL）或序列对象（如Oracle），但Hive中没有这样的机制，必须用其它方法实现。在后面“数据转换与装载”中将说明渐变维的概念和Hive中生成代理键的方法。

1. 选择文件格式

Hive是Hadoop上的数据仓库组件，便于查询和管理分布式存储上的大数据集。Hive提供了一种称为HiveQL的语言，允许用户进行类似于SQL的查询。和普遍使用的所有SQL方言一样，它不完全遵守任何一种ANSI SQL标准，并对标准SQL进行了扩展。HiveQL和MySQL的方言最为接近，但是两者还是存在显著差异。HiveQL只处理结构化数据，并且不区分大小写。缺省时Hive使用内建的derby数据库存储元数据，也可以配置Hive使用MySQL、Oracle等关系数据库存储元数据，生产环境建议使用外部数据库存储Hive元数据。Hive里的数据最终存储在HDFS的文件中，常用的数据文件格式有以下4种：

• TEXTFILE
• SEQUENCEFILE
• RCFILE
• ORCFILE

在深入讨论各种类型的文件格式前，先看一下什么是文件格式。所谓文件格式是一种信息被存储或编码成计算机文件的方式。在Hive中文件格式指的是记录以怎样的编码格式被存储到文件中。当我们处理结构化数据时，每条记录都有自己的结构。记录在文件中是如何编码的就定义了文件格式。不同文件格式的主要区别在于它们的数据编码、压缩率、使用的空间和磁盘I/O。

当用户向传统数据库中增加数据的时候，系统会检查写入的数据与表结构是否匹配，如果不匹配则拒绝插入数据，这就是所谓的写时模式。Hive与此不同，它使用的是读时模式，就是直到读取时再进行数据校验。在向Hive装载数据时，它并不验证数据与表结构是否匹配，但这时它会检查文件格式是否和表定义相匹配。

（1）TEXTFILE

TEXTFILE就是普通的文本型文件，是Hadoop里最常用的输入输出格式，也是Hive的缺省文件格式。如果表定义为TEXTFILE，则可以向该表中装载以逗号、TAB或空格作为分隔符的数据，也可以导入JSON格式的数据。文本文件中除了可以包含普通的字符串、数字、日期等简单数据类型外，还可以包含复杂的集合数据类型。如表4-1所示，Hive支持STRUCT、MAP和ARRAY三种集合数据类型。

表4-1 Hive的集合数据类型

Hive中缺省的记录和字段分隔符如表4-2所示。TEXTFILE格式缺省每一行被认为是一条记录。

表4-2 Hive中缺省的记录和字段分隔

（2）SEQUENCEFILE

我们知道Hadoop处理少量大文件比大量小文件的性能要好。如果文件小于Hadoop里定义的块尺寸（Hadoop 2.x缺省是128M），可以认为是小文件。元数据的增长将转化为NameNode的开销。如果有大量小文件，NameNode会成为性能瓶颈。为了解决这个问题，Hadoop引入了sequence文件，将sequence作为存储小文件的容器。

Sequence文件是由二进制键值对组成的平面文件。Hive将查询转换成MapReduce作业时，决定一个给定记录的哪些键/值对被使用。Sequence文件是可分割的二进制格式，主要的用途是联合多个小文件。

（3）RCFILE

RCFILE指的是Record Columnar File，是一种高压缩率的二进制文件格式，被用于在一个时间点操作多行的场景。RCFILEs是由二进制键/值对组成的平面文件，这点与SEQUENCEFILE非常相似。RCFILE以记录的形式存储表中的列，即列存储方式。它先分割行做水平分区，然后分割列做垂直分区。RCFILE把一行的元数据作为键，把行数据作为值。这种面向列的存储在执行数据分析时更高效。

（4）ORCFILE ORC指的是Optimized Record Columnar，就是说相对于其它文件格式，它以更优化的方式存储数据。ORC能将原始数据的大小缩减75%，从而提升了数据处理的速度。OCR比Text、Sequence和RC文件格式有更好的性能，而且ORC是目前Hive中唯一支持事务的文件格式。

应该依据数据需求选择适当的文件格式，例如：

• 如果数据有参数化的分隔符，那么可以选择TEXTFILE格式。
• 如果数据所在文件比块尺寸小，可以选择SEQUENCEFILE格式。
• 如果想执行数据分析，并高效地存储数据，可以选择RCFILE格式。
• 如果希望减小数据所需的存储空间并提升性能，可以选额ORCFILE格式。

多维数据仓库需要处理渐变维（SCD），必然要用到行级更新，而当前的Hive只有ORCFILE文件格式可以支持此功能。因此在我们的销售订单示例中，所有数据仓库里的表，除日期维度表外，其它表都使用ORCFILE格式。日期维度表数据一旦生成就不会修改，所以使用TEXTFILE格式。原始数据存储里的表数据是从源数据库直接导入的，只有追加和覆盖两种导入方式，不存在数据更新的问题，因此使用缺省的TEXTFILE格式。

2. 选择表类型

（1）管理表 管理表有时也被称为内部表，因为Hive会控制这些表中数据的生命周期。默认情况下，Hive会将这些表的数据存储在由hive-site.xml文件中属性hive.metastore.warehouse.dir所定义目录的子目录下。当我们删除一个管理表时，Hive也会删除这个表中的数据。

管理表的主要问题是只能用Hive访问，不方便和其它系统共享数据。例如，假如有一份由Pig或其它工具创建并且主要由这一工具使用的数据，同时希望使用Hive在这份数据上执行一些查询，可是并没有给予Hive对数据的所有权，这时就不能使用管理表了。我们可以创建一个外部表指向这份数据，而并不需要对其具有所有权。（2）外部表 我们来看一个Hive文档中外部表的例子。

create external table page_view(viewtime int, userid bigint,
     page_url string, referrer_url string,
     ip string comment 'ip address of the user',
     country string comment 'country of origination')
 comment 'this is the staging page view table'
 row format delimited fields terminated by '\054'
 stored as textfile
 location '<hdfs_location>';

上面的语句建立一个名为page_view的外部表。EXTERNAL关键字告诉Hive这是一个外部表，后面的LOCATION子句指示数据位于HDFS的哪个路径下，而不使用hive.metastore.warehouse.dir定义的缺省位置。外部表方便对已有数据的集成。

因为表是外部的，所以Hive并不认为其完全拥有这个表的数据。在对外部表执行删除操作时，只是删除掉描述表的元数据信息，并不会删除表数据。

我们需要清楚的重要一点是管理表和外部表之间的差异要比看起来的小得多。即使对于管理表，用户也可以指定数据是存储在哪个路径下的，因此用户也可以使用其它工具（如hdfs的dfs命令等）来修改甚至删除管理表所在路径下的数据。从严格意义上说，Hive是管理着这些目录和文件，但是并不具有对它们的完全控制权。Hive实际上对于所存储的文件的完整性以及数据内容是否和表结构一致并没有支配能力，甚至管理表都没有给用户提供这些管理能力。 用户可以在DESCRIBE FORMATTED tablename语句的输出中看到表是管理表还是外部表。对于管理表，用户可以看到如下信息：

...
Table Type:             MANAGED_TABLE
...
对于外部表，用户可以看到如下信息：
...
Table Type:             EXTERNAL_TABLE
...

（3）分区表 和其它数据库类似，Hive中也有分区表的概念。分区表的优势体现在可维护性和性能两方面，而且分区表还可以将数据以一种符合业务逻辑的方式进行组织，因此是数据仓库中经常使用的一种技术。管理表和外部表都可以创建相应的分区表，分别称之为管理分区表和外部分区表。

先看一个管理分区表的例子：

create table page_view(viewtime int, userid bigint,
     page_url string, referrer_url string,
     ip string comment 'ip address of the user')
 comment 'this is the page view table'
 partitioned by (dt string, country string)
 row format delimited fields terminated by '\001'
stored as sequencefile;

CREATE TABLE语句的PARTITIONED BY子句用于创建分区表。上面的语句创建一个名为page_view的分区表。这是一个常见的页面浏览记录表，包含浏览时间、浏览用户ID、浏览页面的URL、上一个访问的URL和用户的IP地址五个字段。该表以日期和国家作为分区字段，存储为SEQUENCEFILE文件格式。文件中的数据分别使用缺省的Ctrl-A和换行符作为列和行的分隔符。 DESCRIBE FORMATTED命令会显示出分区键：

hive> DESCRIBE FORMATTED page_view;
# col_name                data_type               comment             
          
viewtime                int                                         
userid                  bigint                                      
page_url                string                                      
referrer_url            string                                      
ip                      string                  IP Address of the User
          
# Partition Information          
# col_name                data_type               comment             
          
dt                      string                                      
country                 string

输出信息中把表字段和分区字段分开显示。这两个分区键当前的注释都是空，我们也可以像给普通字段增加注释一样给分区字段增加注释。

分区表改变了Hive对数据存储的组织方式。如果是一个非分区表，那么只会有一个page_view目录与之对应，而对于分区表，当向表中装载数据后，Hive将会创建好可以反映分区结构的子目录。分区字段一旦创建好，表现得就和普通字段一样。事实上，除非需要优化查询性能，否则用户不需要关心字段是否是分区字段。需要注意的是，通常分区字段的值包含在目录名称中，而不在它们目录下的文件中。也有分区字段的值不包含在目录名称中的情况。

对数据进行分区，最重要的原因就是为了更快地查询。如果用户的查询包含“where dt = '...' and country = '...'”这样的条件，查询优化器只需要扫描一个分区目录即可。即使有很多日期和国家的目录，除了一个目录其它的都可以忽略不计，这就是所谓的“分区消除”。对于非常大的数据集，利用分区消除特性可以显著地提高查询性能。当我们在WHERE子句中增加谓词来按照分区值进行过滤时，这些谓词被称为分区过滤器。

当然，如果用户需要做一个查询，查询中不带分区过滤器，甚至查询的是表中的全部数据，那么Hive不得不读取表目录下的每个子目录，这种宽范围的磁盘扫描是应该尽量避免的。如果表中的数据以及分区个数都非常大的话，执行这样一个包含所有分区的查询可能会触发一个巨大的MapReduce任务。一个强烈建议的安全措施是将Hive设置为严格mapred模式，这样如果对分区表进行查询而WHERE子句没有加分区过滤的话，将会禁止提交这个查询。

外部表同样可以使用分区，事实上，这是管理大型生产数据集最常见的情况。这种结合给用户提供了一个可以和其它工具共享数据的方式，同时也可以优化查询性能。由于用户能够自己定义目录结构，因此用户对于目录结构的使用具有更多的灵活性。日志文件分析就非常适合这种场景。

例如我们有一个用户下载手机APP的日志文件，其中记录了手机操作系统、下载时间、下载渠道、下载的APP、下载用户和其它杂项信息，杂项信息使用一个JSON字符串表示。应用程序每天会生成一个新的日志文件。我们可以按照如下方式来定义对应的Hive表：

create external table logs(
platform          string,
    createtime        string,
    channel            string,
    product            string,  
    userid            string,
    content            map<string,string>)
partitioned by (dt int)
row format delimited fields terminated by '\t'
location 'hdfs://cdh2/logs';

我们建立了一个外部分区表，dt是分区字段，它是日期的整数表示。将日志数据按天进行分区，划分的数据量大小合适，而且按天这个粒度进行查询也能满足需求。每天定时执行以下的shell脚本，把前一天生成的日志文件装载进Hive。脚本执行后，就可以使用Hive表分析前一天的日志数据了。脚本中使用hive命令行工具的-e参数执行HiveQL语句。

#!/bin/bash
# 设置环境变量
source /home/work/.bash_profile
# 取得前一天的日期，格式为yyyymmdd，作为分区的目录名
dt=$(date -d last-day +%Y%m%d)
# 建立HDFS目录
hadoop fs -mkdir -p /logs/$dt
# 将前一天的日志文件上传到HDFS的相应目录中
hadoop fs -put /data/statsvr/tmp/logs_$dt /logs/$dt
# 给Hive表增加一个新的分区，指向刚建的目录
hive --database logs -e "alter table logs add partition(dt=$dt) location 'hdfs://cdh2/logs/$dt'"

Hive并不关心一个分区对应的分区目录是否存在或者分区目录下是否有文件。如果分区目录不存在或分区目录下没有文件，则对于这个分区的查询将没有返回结果。当用户想在另外一个进程开始往分区中写数据之前创建好分区时，这样处理是很方便的。数据一旦存在，对它的查询就会有返回结果。

这个功能所具有的另一个好处是，可以将新数据写入到一个专用的目录中，并与位于其它目录中的数据存在明显的区别。不管用户是将旧数据转移到一个归档位置还是直接删除掉，新数据被篡改和误删除的风险被降低了，因为新数据位于不同的目录下。和非分区外部表一样，Hive并不控制数据，即使表被删除，数据也不会被删除。

本示例中的Hive表均为普通非分区管理表，这出于两点考虑。一是示例目的是说明Hive可以满足建设传统多维数据仓库的技术要求，而不是展示它的全部特性。二是本示例更像是一个POC验证，我们尽量简化用例，不过多涉及性能优化、缓存、安全或其它复杂主题。

3. 支持行级更新

HDFS是一个不可更新的文件系统，其中只能创建、删除文件或目录，文件一旦创建，只能从它的末尾追加数据，已存在数据不能修改。Hive以HDFS为基础，Hive表里的数据最终会物理存储在HDFS上，因此原生的Hive是不支持insert ... values、update、delete等事务处理或行级更新的。这种情况直到Hive 0.14才有所改变。该版本具有一定的事务处理能力，在此基础上支持行级数据更新。

为了在HDFS上支持事务，Hive将表或分区的数据存储在基础文件中，而将新增的、修改的、删除的记录存储在一种称为delta的文件中。每个事务都将产生一系列delta文件。在读取数据时Hive合并基础文件和delta文件，把更新或删除操作应用到基础文件中。

Hive已经支持完整ACID特性的事务语义，因此功能得到了扩展，增加了以下使用场景：

• 获取数据流。很多用户在Hadoop集群中使用了诸如Apache Flume、Apache Storm或者Apache Kafka进行流数据处理。这些工具每秒可能写数百行甚至更多的数据。在支持事务以前，Hive只能通过增加分区的方式接收流数据。通常每隔15分钟到1小时新建一个分区，快速的数据载入会导致表中产生大量的分区。这种方案有两个个明显的问题。一是当前述的流数据处理工具向已存在的分区中装载数据时，可能会对正在读取数据的用户产生脏读，也就是说，用户可能读取到他们在开始查询时间点后写入的数据。二是会在表目录中遗留大量的小数据文件，这将给NameNode造成很大压力。支持事务功能后，应用就可以向Hive表中持续插入数据行，避免产生太多的文件，并且向用户提供数据的一致性读。
• 处理渐变维（SCD）。在一个典型的星型模式数据仓库中，维度表随时间的变化很缓慢。例如，一个零售商开了一家新商店，需要将新店数据加到商店表，或者一个已有商店的营业面积或其它需要跟踪的特性改变了。这些改变会导致插入或修改个别记录（依赖于选择的策略）。从0.14版开始，Hive支持了事务及行级更新，从而能够处理各种SCD类型。
• 数据修正。有时候我们需要修改已有的数据。如先前收集的数据是错误的，或者第一次得到的可能只是部分数据（例如90%的服务器报告），而完整的数据会在后面提供，或者业务规则可能要求某些事务因为后续事务而重新启动，（例如，一个客户购买了商品后，又购买了一张会员卡，因此获得了包括之前所购买商品在内的折扣价格。）或者在合作关系结束后，依据合同需要删除客户的数据等等。这些数据处理都需要执行insert、update或delete操作。

Hive 0.14后开始支持事务，但缺省是不支持的，需要一些附加的配置。本示例环境CDH 6.3.1包含的Hive版本是2.1.1，该版本可以支持事务及行级更新。

要让Hive支持行级更新，需要进行一点配置，并且在建表时，必须指定存储格式为ORC，并且必须分桶，而且在表属性中必须指定transaction=true。下面演示如何设置hive表支持insert、update、delete操作。

首先在hive配置文件中增加属性hive.txn.manager，值为org.apache.hadoop.hive.ql.lockmgr.DbTxnManager。在Cloudera Manager中，选择Hive服务，在“配置”标签搜索“hive-site.xml”。然后在“hive-site.xml 的 Hive 服务高级配置代码段（安全阀）”和“hive-site.xml 的 Hive 客户端高级配置代码段（安全阀）”中添加该属性值，如图4-3所示。保存修改后，重启Hive服务。

图4-3 增加hive.txn.manager属性

hive.txn.manager的缺省值为org.apache.hadoop.hive.ql.lockmgr.DummyTxnManager，有org.apache.hadoop.hive.ql.lockmgr.DummyTxnManager和org.apache.hadoop.hive.ql.lockmgr.DbTxnManager两种取值。前者是Hive 0.13之前版本的锁管理器，不提供事务支持。后者是Hive 0.13.0版本为了支持事务新加的属性值。

最后测试Hive的行级更新，下面这些SQL语句都可以正常执行：

use test;

create table t_update(id int, name string)     
clustered by (id) into 8 buckets     
stored as orc tblproperties ('transactional'='true'); 

insert into t_update values (1,'aaa'), (2,'bbb');

update t_update set name='ccc' where id=1;

delete from t_update where id=2;

select * from t_update;

说明：

• 必须存储为ORC格式。
• 建表语句必须带有into buckets子句和stored as orc tblproperties ('transactional'='true')子句，并且不能带有sorted by子句。
• 关键字clustered声明划分桶的列和桶的个数，这里以id来划分桶，划分8个桶。Hive会计算id列的hash值再以桶的个数取模来计算某条记录属于哪个桶。

4. Hive事务支持的限制

现在的Hive虽然已经支持了事务，但是并不完善，存在很多限制。我们还不能象使用关系数据库那样来操作Hive，这是由MapReduce计算框架和CAP理论所决定的。Hive事务处理的局限性体现在以下几个方面。

• 暂不支持BEGIN、COMMIT和ROLLBACK语句，所有HiveQL语句都是自动提交的。Hive计划在未来版本支持这些语句。
• 现有版本只支持ORC文件格式，未来可能会支持所有存储格式。Hive计划给表中的每行记录增加显式或隐式的row id，用于行级的update或delete操作。这项功能很值得期待，但目前来看进展不大。
• 缺省配置下，事务功能是关闭的，必须进行一些配置才能使用事务，易用性不理想。
• 使用事务的表必须分桶，而相同系统上不使用事务和ACID特性的表则没有此限制。
• 外部表的事务特性有可能失效。
• 不允许从一个非ACID的会话读写事务表。换句话说，会话中的锁管理器变量必须设置成org.apache.hadoop.hive.ql.lockmgr.DbTxnManager，才能与事务表一起工作。
• 当前版本只支持快照级别的事务隔离。当一个查询开始执行后，Hive提供给它一个查询开始时间点的数据一致性快照。传统事务的脏读、读提交、可重复读或串行化隔离级别都不支持。计划引入的BEGIN语句，目的就是在事务执行期间支持快照隔离级别，而不仅仅是面向单一语句。Hive官方称会依赖用户需求增加其它隔离级别。
• ZooKeeper和内存锁管理器与事务不兼容。

三、建立数据库表

现在我们已经清楚了Hive支持的文件格式和表类型，以及如何支持事务和装载数据等问题，下面就来创建开篇说明的销售订单数据仓库中的表。在这个场景中，源数据库表就是操作型系统的模拟。我们在MySQL中建立源数据库表。RDS存储原始数据，作为源数据到数据仓库的过渡，在Hive中建RDS库表。TDS即为转化后的多维数据仓库，在Hive中建TDS库表。

1. 源数据库表

执行下面的SQL语句在MySQL中建立源数据库表。

-- 建立源数据库  
drop database if exists source;  
create database source;  

use source;  
-- 建立客户表    
create table customer (    
    customer_number int not null auto_increment primary key comment '客户编号，主键',    
    customer_name varchar(50) comment '客户名称',    
    customer_street_address varchar(50) comment '客户住址',    
    customer_zip_code int comment '邮编',    
    customer_city varchar(30) comment '所在城市',    
    customer_state varchar(2) comment '所在省份'    
);  
-- 建立产品表    
create table product (       
    product_code int not null auto_increment primary key comment '产品编码，主键',    
    product_name varchar(30) comment '产品名称',    
    product_category varchar(30) comment '产品类型'       
);  
-- 建立销售订单表    
create table sales_order (    
    order_number int not null auto_increment primary key comment '订单号，主键',    
    customer_number int comment '客户编号',    
    product_code int comment '产品编码',    
    order_date datetime comment '订单日期',    
    entry_date datetime comment '登记日期',    
    order_amount decimal(10 , 2 ) comment '销售金额',    
    foreign key (customer_number)    
        references customer (customer_number)    
        on delete cascade on update cascade,    
    foreign key (product_code)    
        references product (product_code)    
        on delete cascade on update cascade    
);

执行下面的SQL语句生成源库测试数据。

use source;  
-- 生成客户表测试数据    
insert into customer 
(customer_name,customer_street_address,customer_zip_code,
customer_city,customer_state) 
values      
('really large customers', '7500 louise dr.',17050, 'mechanicsburg','pa'),    
('small stores', '2500 woodland st.',17055, 'pittsburgh','pa'),    
('medium retailers','1111 ritter rd.',17055,'pittsburgh','pa'),    
('good companies','9500 scott st.',17050,'mechanicsburg','pa'),    
('wonderful shops','3333 rossmoyne rd.',17050,'mechanicsburg','pa'),    
('loyal clients','7070 ritter rd.',17055,'pittsburgh','pa'),    
('distinguished partners','9999 scott st.',17050,'mechanicsburg','pa');  

-- 生成产品表测试数据    
insert into product (product_name,product_category)    
values     
('hard disk drive', 'storage'),    
('floppy drive', 'storage'),    
('lcd panel', 'monitor');  
  
-- 生成100条销售订单表测试数据  
drop procedure if exists generate_sales_order_data;  
delimiter //    
create procedure generate_sales_order_data()  
begin   
    drop table if exists temp_sales_order_data;  
    create table temp_sales_order_data as select * from sales_order where 1=0;  
  
    set @start_date := unix_timestamp('2020-03-01');  
    set @end_date := unix_timestamp('2020-09-01');  
    set @i := 1;  
  
    while @i<=100 do  
        set @customer_number := floor(1 + rand() * 6);  
        set @product_code := floor(1 + rand() * 2);  
        set @order_date := from_unixtime(@start_date + rand() * (@end_date - @start_date));  
        set @amount := floor(1000 + rand() * 9000);  
  
        insert into temp_sales_order_data values (@i,@customer_number,@product_code,@order_date,@order_date,@amount);  
        set @i:=@i+1;  
    end while;  
  
    truncate table sales_order;  
    insert into sales_order   
    select null,customer_number,product_code,order_date,entry_date,order_amount from temp_sales_order_data order by order_date;  
    commit;  
  
end   
//    
delimiter ;   
  
call generate_sales_order_data(); 

说明：

• 客户表和产品表的测试数据取自《Dimensional Data Warehousing with MySQL》一书。
• 创建了一个MySQL存储过程生成100条销售订单测试数据。为了模拟实际订单的情况，订单表中的客户编号、产品编号、订单时间和订单金额都取一个范围内的随机值，订单时间与登记时间相同。因为订单表的主键是自增的，为了保持主键值和订单时间字段的值顺序保持一致，引入了一个名为temp_sales_order_data的表，存储中间临时数据。在本专题后面示例中都是使用此方案生成订单测试数据。

2. RDS库表

执行下面的HiveQL语句在Hive中建立RDS库表。

-- 建立rds数据库    
drop database if exists rds cascade;  
create database rds;  
  
use rds;  
-- 建立客户过渡表    
create table customer (    
    customer_number int comment '客户编号',  
    customer_name varchar(30) comment '客户名称',  
    customer_street_address varchar(30) comment '客户住址',  
    customer_zip_code int comment '邮编',  
    customer_city varchar(30) comment '所在城市',  
    customer_state varchar(2) comment '所在省份'  
);  
-- 建立产品过渡表    
create table product (    
    product_code int comment '产品编码',    
    product_name varchar(30) comment '产品名称',    
    product_category varchar(30) comment '产品类型'    
);  
-- 建立销售订单过渡表  
create table sales_order (    
    order_number int comment '订单号',    
    customer_number int comment '客户编号',    
    product_code int comment '产品编码',    
    order_date timestamp comment '订单日期',    
    entry_date timestamp comment '登记日期',    
    order_amount decimal(10 , 2 ) comment '销售金额'  
);  

说明：

• RDS中表与MySQL里的源表完全对应，其字段与源表相同。
• 使用Hive缺省的文件格式。
• 为了使Hive能够正常显示中文注释，需要修改存储Hive元数据的相关列的字符集。本例使用MySQL作为Hive元数据存储数据库，因此执行下面的SQL语句，修改三个Hive元数据表列的字符集，由原来的latin1改为utf8，然后再建表即可。 alter table COLUMNS_V2 modify column COMMENT varchar(256) character set utf8; alter table TABLE_PARAMS modify column PARAM_VALUE varchar(4000) character set utf8; alter table PARTITION_KEYS modify column PKEY_COMMENT varchar(4000) character set utf8;

3. TDS库表

执行下面的HiveQL语句在Hive中建立TDS库表。

-- 建立数据仓库数据库    
drop database if exists dw cascade;    
create database dw;  

use dw;  
-- 建立日期维度表  
create table date_dim (    
    date_sk int comment '日期代理键',  
    dt date comment '日期，格式为yyyy-mm-dd',  
    month tinyint comment '月份',  
    month_name varchar(9) comment '月名称',  
    quarter tinyint comment '季度',  
    year smallint comment '年份'  
)  
comment '日期维度表'  
row format delimited fields terminated by ','  
stored as textfile; 

-- 建立客户维度表    
create table customer_dim (    
    customer_sk int comment '代理键',    
    customer_number int comment '客户编号，业务主键',    
    customer_name varchar(50) comment '客户名称',    
    customer_street_address varchar(50) comment '客户住址',    
    customer_zip_code int comment '邮编',    
    customer_city varchar(30) comment '所在城市',    
    customer_state varchar(2) comment '所在省份',  
    version int comment '版本号',  
    effective_date date comment '生效日期',    
    expiry_date date comment '到期日期'    
)  
clustered by (customer_sk) into 8 buckets    
stored as orc tblproperties ('transactional'='true');  
    
-- 建立产品维度表    
create table product_dim (    
    product_sk int comment '代理键',    
    product_code int comment '产品编码，业务主键',    
    product_name varchar(30) comment '产品名称',    
    product_category varchar(30) comment '产品类型',  
    version int comment '版本号',  
    effective_date date comment '生效日期',    
    expiry_date date comment '到期日期'    
)  
clustered by (product_sk) into 8 buckets    
stored as orc tblproperties ('transactional'='true');  
    
-- 建立订单维度表    
create table order_dim (    
    order_sk int comment '代理键',    
    order_number int comment '订单号，业务主键',   
    version int comment '版本号',  
    effective_date date comment '生效日期',  
    expiry_date date comment '到期日期'    
)  
clustered by (order_sk) into 8 buckets    
stored as orc tblproperties ('transactional'='true');  
    
-- 建立销售订单事实表    
create table sales_order_fact (    
    order_sk int comment '订单维度代理键',    
    customer_sk int comment '客户维度代理键',    
    product_sk int comment '产品维度代理键',    
    order_date_sk int comment '日期维度代理键',    
    order_amount decimal(10 , 2 ) comment '销售金额' 
)  
clustered by (order_sk) into 8 buckets    
stored as orc tblproperties ('transactional'='true');  

说明：

• 按照图4-2所示的实体关系建立多维数据仓库中的维度表和事实表。
• 除日期维度表外，其它表都使用ORC文件格式，并设置表属性支持事务。
• 日期维度表只会追加数据而从不更新，所以使用以逗号作为列分隔符的文本文件格式。
• 维度表虽然使用了代理键，但不能将它设置为主键，在数据库级也不能确保其唯一性。Hive中并没有主键、外键、唯一性约束、非空约束这些关系数据库的概念。

四、装载日期维度数据

日期维度在数据仓库中是一个特殊角色。日期维度包含时间概念，而时间是最重要的，因为数据仓库的主要功能之一就是存储历史数据，所以每个数据仓库里的数据都有一个时间特征。装载日期数据有三个常用方法：预装载、每日装载一天、从源数据装载日期。

在三种方法中，预装载最为常见也最容易实现，本示例就采用此方法，生成一个时间段里的所有日期。我们预装载5年的日期维度数据，从2018年1月1日到2022年12月31日。使用这个方法，在数据仓库生命周期中，只需要预装载日期维度一次。预装载的缺点是：提早消耗磁盘空间（这点空间占用通常是可以忽略的）；可能不需要所有的日期（稀疏使用）。

下面新建一个如图4-4所示的Kettle转换生成日期维度表数据。

图4-4 生成日期维度表数据的转换

该转换包括四个步骤。第一个步骤是“生成记录”，编辑该步骤属性如下：

• 限制：1826。该属性决定生成的记录数。我们要生成从2018年1月1日到2022年12月31日五年的日期，共1826天，所以这里输入1826。
• 字段：如表4-3所示。

表4-3 生成记录字段

language_code定义语言编码，country_code定义国家编码，这两个字段在“JavaScript代码”中要用到。initial_date字段定义初始日期值。该步骤的输出是1826个同样的行：

en    ca    2018-01-01

第二个步骤是“增加序列”，编辑该步骤属性如下：

• 值的名称：输入“DaySequence”。
• 使用计数器计算sequence：勾选。

该步骤为前一步骤的每行生成一个序号列，输出如下的1826行4列，第4列的列名是“DaySequence”。

en    ca    2018-01-01    1
en    ca    2018-01-01    2
...
en    ca    2018-01-01    1825
en    ca    2018-01-01    1826

第三个步骤是“JavaScript代码”，编辑该步骤属性如下：

• Java Script：代码如下。 //Create a Locale according to the specified language code var locale = new java.util.Locale( language_code.getString() , country_code.getString() ); //Create a calendar, use the specified initial date var calendar = new java.util.GregorianCalendar(locale); calendar.setTime(initial_date.getDate()); //set the calendar to the current date by adding DaySequence days calendar.add(calendar.DAY_OF_MONTH,DaySequence.getInteger() - 1); var simpleDateFormat = java.text.SimpleDateFormat("D",locale); //get the calendar date var dt = new java.util.Date(calendar.getTimeInMillis()); simpleDateFormat.applyPattern("MM"); var month_number = simpleDateFormat.format(dt); simpleDateFormat.applyPattern("MMMM"); var month_name = simpleDateFormat.format(dt); simpleDateFormat.applyPattern("yyyy"); var year4 = "" + simpleDateFormat.format(dt); var quarter_number; switch(parseInt(month_number)){ case 1: case 2: case 3: quarter_number = "1"; break; case 4: case 5: case 6: quarter_number = "2"; break; case 7: case 8: case 9: quarter_number = "3"; break; case 10: case 11: case 12: quarter_number = "4"; break; } var date_key = DaySequence;
• 兼容模式？：勾选。
• 字段：如表4-4所示。

表4-4 JavaScript输出的字段

JavaScript代码输出日期代理键、日期、月份、月份名称、季度、年份6个字段。代理键取的就是前一步骤输出的DaySequence字段的值。日期以initial_date的值加上（DaySequence-1）天生成。用simpleDateFormat生成月份、月份名称、季度、年份的值。

第四个步骤是“Hadoop file output”，编辑该步骤属性如下： （1）“文件”标签

• Hadoop Cluster：选择“CDH631”。
• Folder/File：输入“/user/hive/warehouse/dw.db/date_dim/date_dim.csv”。该路径是date_dim表所对应的HDFS路径。

其它都为空。

（2）“内容”标签

• 分隔符：输入“,”。这是我们在创建date_dim表时选择的文本文件列分隔符。
• 封闭符：空。
• 头部：勾掉。
• 格式：选择“LF terminated(Unix)”。
• 编码：选择“UTF-8”。

（3）“字段”标签 输入如表4-5所示。

表4-5 date_dim.csv文件对应的字段

该步骤将前面步骤的输出传输到HDFS的date_dim.csv文件中。这里不要使用“表输出”步骤向Hive表插入数据。虽然我们配置了Hive支持行级插入，但逐行向Hive表insert数据的速度慢到令人无法忍受。保存并执行转换，HDFS上生成的文件如下：

[root@node3~]#hdfs dfs -ls /user/hive/warehouse/dw.db/date_dim/
Found 1 items
-rw-r--r--   3 root hive      58057 2020-09-04 17:24 /user/hive/warehouse/dw.db/date_dim/date_dim.csv
[root@node3~]#

查询date_dim表结果如下：

hive> select count(*) from dw.date_dim;
...

OK
1826

hive> select * from dw.date_dim order by dt limit 5;
...
OK
1    2018-01-01    1    January    1    2018
2    2018-01-02    1    January    1    2018
3    2018-01-03    1    January    1    2018
4    2018-01-04    1    January    1    2018
5    2018-01-05    1    January    1    2018

hive> select * from dw.date_dim order by dt desc limit 5;
...
OK
1826    2022-12-31    12    December    4    2022
1825    2022-12-30    12    December    4    2022
1824    2022-12-29    12    December    4    2022
1823    2022-12-28    12    December    4    2022
1822    2022-12-27    12    December    4    2022

该转换可以重复执行多次，每次执行结果是相同的，即实现了所谓的“幂等操作”。至此，我们的示例数据仓库模型搭建完成，后面在其上将实现ETL。

五、小节

我们使用一个简单而典型的销售订单示例，建立数据仓库模型。Hive常用的四种文件格式为TEXTFILE、SEQUENCEFILE、RCFILE、ORCFILE，其中只有ORCFILE支持事务和行级更新，因此是多维数据仓库Hive存储类型的唯一选择。Hive中的表分为管理表和外部表，两者都可以进行分区。配置Hive支持事务需要在hive-site.xml文件中增加hive.txn.manager属性。本示例模型在MySQL中建立源库表，在Hive中建立RDS和TDS库表。最后用一个Kettle转换预装载日期维度表数据。