HAWQ取代传统数仓实践（三）——初始ETL（Sqoop、HAWQ）

一、用sqoop用户建立初始抽取脚本

        本示例要用Sqoop将MySQL的数据抽取到HDFS上的指定目录，然后利用HAWQ外部表功能将HDFS数据文件装载到内部表中。表1汇总了示例中维度表和事实表用到的源数据表及其抽取模式。
[align=center]

源数据表	HDFS目录	对应EXT模式中的表	抽取模式
customer	/data/ext/customer	customer	整体、拉取
product	/data/ext/product	product	整体、拉取
sales_order	/data/ext/sales_order	sales_order	基于时间戳的CDC、拉取

[/align]表1

1. 覆盖导入 

        对于customer、product这两个表采用整体拉取的方式抽数据。ETL通常是按一个固定的时间间隔，周期性定时执行的，因此对于整体拉取的方式而言，每次导入的数据需要覆盖上次导入的数据。Sqoop提供了delete-target-dir参数实现覆盖导入。该参数指示在每次抽取数据前先将目标目录删除，作用是提供了一个幂等操作的选择。所谓幂等操作指的是其执行任意多次所产生的影响均与一次执行的影响相同。这样就能在导入失败或修复bug后可以再次执行该操作，而不用担心重复执行会对系统造成数据混乱。

2. 增量导入

        Sqoop提供增量导入模式，用于只导入比已经导入行新的数据行。表2所示参数用来控制增量导入。

[align=center]

参数	描述
--check-column	在确定应该导入哪些行时，指定被检查的列。列不能是CHAR/NCHAR/VARCHAR/VARNCHAR/LONGVARCHAR/LONGNVARCHAR数据类型。
--incremental	指定Sqoop怎样确定哪些行是新行。有效值是append和lastmodified。
--last-value	指定已经导入数据的被检查列的最大值。

[/align]表2
        Sqoop支持两种类型的增量导入：append和lastmodified。可以使用--incremental参数指定增量导入的类型。
        当被导入表的新行具有持续递增的行id值时，应该使用append模式。指定行id为--check-column的列。Sqoop导入那些被检查列的值比--last-value给出的值大的数据行。
        Sqoop支持的另一个表修改策略叫做lastmodified模式。当源表的数据行可能被修改，并且每次修改都会更新一个last-modified列为当前时间戳时，应该使用lastmodified模式。那些被检查列的时间戳比last-value给出的时间戳新的数据行被导入。
        增量导入命令执行后，在控制台输出的最后部分，会打印出后续导入需要使用的last-value。当周期性执行导入时，应该用这种方式指定--last-value参数的值，以确保只导入新的或修改过的数据。可以通过一个增量导入的保存作业自动执行这个过程，这是适合重复执行增量导入的方式。
        有了对Sqoop增量导入的基本了解，下面看一下如何在本示例中使用它抽取数据。对于sales_order这个表采用基于时间戳的CDC拉取方式抽数据。这里假设源系统中销售订单记录一旦入库就不再改变，或者可以忽略改变。也就是说销售订单是一个随时间变化单向追加数据的表。sales_order表中有两个关于时间的字段，order_date表示订单时间，entry_date表示订单数据实际插入表里的时间，两个时间可能不同。那么用哪个字段作为CDC的时间戳呢？设想这样的情况，一个销售订单的订单时间是2017年1月1日，实际插入表里的时间是2017年1月2日，ETL每天0点执行，抽取前一天的数据。如果按order_date抽取数据，条件为where order_date >= '2017-01-02' AND order_date < '2017-01-03'，则2017年1月3日0点执行的ETL不会捕获到这个新增的订单数据。所以应该以entry_date作为CDC的时间戳。

3. 编写初始数据抽取脚本

        用sqoop操作系统用户建立初始数据抽取脚本文件~/init_extract.sh，内容如下：

#!/bin/bash

# 建立Sqoop增量导入作业，以order_number作为检查列，初始的last-value是0
sqoop job --delete myjob_incremental_import
sqoop job --create myjob_incremental_import \
-- import \
--connect "jdbc:mysql://172.16.1.127:3306/source?usessl=false&user=dwtest&password=123456" \
--table sales_order \
--target-dir /data/ext/sales_order \
--compress \
--where "entry_date < current_date()" \
--incremental append \
--check-column order_number \
--last-value 0

# 全量抽取客户表
sqoop import --connect jdbc:mysql://172.16.1.127:3306/source --username dwtest --password 123456 --table customer --targe
t-dir /data/ext/customer --delete-target-dir --compress

# 全量抽取产品表
sqoop import --connect jdbc:mysql://172.16.1.127:3306/source --username dwtest --password 123456 --table product --target
-dir /data/ext/product --delete-target-dir --compress

# 首次全量抽取销售订单表
sqoop job --exec myjob_incremental_import

        说明：
为了保证外部表数据量尽可能小，使用compress选项进行压缩，Sqoop缺省的压缩算法是gzip，hdfstextsimples属性的HAWQ PXF外部表能自动正确读取这种格式的压缩文件。
执行时先重建Sqoop增量抽取作业，指定last-value为0。由于order_number都是大于0的，因此初始时会装载所有订单数据。
        将文件修改为可执行模式：

chmod 755 ~/init_extract.sh

二、用gpadmin用户建立初始装载脚本

        在数据仓库可以使用前，需要装载历史数据。这些历史数据是导入进数据仓库的第一个数据集合。首次装载被称为初始装载，一般是一次性工作。由最终用户来决定有多少历史数据进入数据仓库。例如，数据仓库使用的开始时间是2017年3月1日，而用户希望装载两年的历史数据，那么应该初始装载2015年3月1日到2017年2月28日之间的源数据。在2017年3月2日装载2017年3月1日的数据（假设执行频率是每天一次），之后周期性地每天装载前一天的数据。在装载事实表前，必须先装载所有的维度表。因为事实表需要引用维度的代理键。这不仅针对初始装载，也针对定期装载。

1. 数据源映射

        表3显示了本示例需要的源数据的关键信息，包括源数据表、对应的数据仓库目标表等属性。客户和产品的源数据直接与其数据仓库里的目标表，customer_dim和product_dim表相对应，而销售订单事务表是多个数据仓库表的数据源。
[align=center]

源数据	源数据类型	文件名/表名	数据仓库中的目标表
客户	MySQL表	customer	customer_dim
产品	MySQL表	product	product_dim
销售订单	MySQL表	sales_order	order_dim、sales_order_fact

[/align]表3

2. 确定SCD处理方法

        标识出了数据源，现在要考虑维度历史的处理。渐变维（SCD）即是一种在多维数据仓库中实现维度历史的技术。有三种不同的SCD技术：SCD 类型1（SCD1），SCD类型2（SCD2），SCD类型3（SCD3）：
SCD1 - 通过更新维度记录直接覆盖已存在的值，它不维护记录的历史。SCD1一般用于修改错误的数据。
SCD2 - 在源数据发生变化时，给维度记录建立一个新的“版本”记录，从而维护维度历史。SCD2不删除、修改已存在的数据。
SCD3 – 通常用作保持维度记录的几个版本。它通过给某个数据单元增加多个列来维护历史。例如，为了记录客户地址的变化，customer_dim维度表有一个customer_address列和一个previous_customer_address列，分别记录当前和上一个版本的地址。SCD3可以有效维护有限的历史，而不像SCD2那样保存全部历史。SCD3很少使用。它只适用于数据的存储空间不足并且用户接受有限维度历史的情况。
        同一个维度表中的不同字段可以有不同的变化处理方式。在传统数据仓库中，对于SCD1一般就直接UPDATE更新属性，而SCD2则要新增记录。但HAWQ没有提供UPDATE、DELETE等DML操作，因此对于所有属性的变化均增加一条记录，即所有维度属性都按SCD2方式处理。

3. 实现代理键

        多维数据仓库中的维度表和事实表一般都需要有一个代理键，作为这些表的主键，代理键一般由单列的自增数字序列构成。HAWQ中的bigserial数据类型与MySQL的auto_increment类似，长用于定义自增列。但它的实现方法却与Oracle的sequence类似，当创建bigserial字段的表时，HAWQ会自动创建一个自增的sequence对象，bigserial字段自动引用sequence实现自增。

4. 编写初始数据装载脚本

        所有技术实现的细节都清楚后，现在编写初始数据装载脚本。需要执行两步主要操作，一是将外部表的数据装载到RDS模式的表中，二是向TDS模式中的表装载数据。用gpadmin操作系统用户建立初始数据装载脚本文件~/init_load.sql，内容如下：

-- 分析外部表
analyze ext.customer;
analyze ext.product;
analyze ext.sales_order;

-- 将外部数据装载到原始数据表
set search_path to rds;

truncate table customer;
truncate table product;
truncate table sales_order;

insert into customer select * from ext.customer;
insert into product select * from ext.product;
insert into sales_order select * from ext.sales_order;

-- 分析rds模式的表
analyze rds.customer;
analyze rds.product;
analyze rds.sales_order;

-- 装载数据仓库数据
set search_path to tds;

truncate table customer_dim;
truncate table product_dim;
truncate table order_dim;
truncate table sales_order_fact;

-- 序列初始化
alter sequence customer_dim_customer_sk_seq restart with 1;
alter sequence product_dim_product_sk_seq restart with 1;
alter sequence order_dim_order_sk_seq restart with 1;

-- 装载客户维度表
insert into customer_dim
(customer_number,
customer_name,
customer_street_address,
customer_zip_code,
customer_city,
customer_state,
version,
effective_date)
select t1.customer_number,
t1.customer_name,
t1.customer_street_address,
t1.customer_zip_code,
t1.customer_city,
t1.customer_state,
1,
'2016-03-01'
from rds.customer t1
order by t1.customer_number;

-- 装载产品维度表
insert into product_dim
(product_code,
product_name,
product_category,
version,
effective_date)
select product_code,
product_name,
product_category,
1,
'2016-03-01'
from rds.product t1
order by t1.product_code;

-- 装载订单维度表
insert into order_dim (order_number,version,effective_date)
select order_number, 1, order_date
from rds.sales_order t1
order by t1.order_number;

-- 装载销售订单事实表
insert into sales_order_fact
select order_sk,
customer_sk,
product_sk,
date_sk,
e.year*100 + e.month,
order_amount
from rds.sales_order a,
order_dim b,
customer_dim c,
product_dim d,
date_dim e
where a.order_number = b.order_number
and a.customer_number = c.customer_number
and a.product_code = d.product_code
and date(a.order_date) = e.date;

-- 分析tds模式的表
analyze customer_dim;
analyze product_dim;
analyze order_dim;
analyze sales_order_fact;

        说明：
装载前清空表、以及重新初始化序列的目的是为了可重复执行初始装载脚本。
依据HAWQ的建议，装载数据后，执行查询前，先分析表以提高查询性能。

三、用root用户建立初始ETL脚本

        前面的数据抽取脚本文件的属主是sqoop用户，而数据装载脚本文件的属主是gpadmin用户。除了这两个用户以外，还需要使用hdfs用户执行文件操作。为了简化多用户调用执行，用root用户将所有需要的操作封装到一个文件中，提供统一的初始数据装载执行入口。
        用root操作系统用户建立初始ETL脚本文件~/init_etl.sh，内容如下：

#!/bin/bash

# 为了可以重复执行初始装载过程，先使用hdfs用户删除销售订单外部表目录
su - hdfs -c 'hdfs dfs -rm -r /data/ext/sales_order/*'

# 使用sqoop用户执行初始抽取脚本
su - sqoop -c '~/init_extract.sh'

# 使用gpadmin用户执行初始装载脚本
su - gpadmin -c 'export PGPASSWORD=123456;psql -U dwtest -d dw -h hdp3 -f ~/init_load.sql'

说明：
Sqoop中incremental append与delete-target-dir参数不能同时使用。因此为了可重复执行Sqoop增量抽取作业，先要用hdfs用户删除相应目录下的所有文件。
使用su命令，以不同用户执行相应的脚本文件。
        将文件修改为可执行模式：

chmod 755 ~/init_etl.sh

四、用root用户执行初始ETL脚本

~/init_etl.sh

        执行以下查询验证初始ETL结果：

select order_number,
customer_name,
product_name,
date,
order_amount amount
from sales_order_fact a,
customer_dim b,
product_dim c,
order_dim d,
date_dim e
where a.customer_sk = b.customer_sk
and a.product_sk = c.product_sk
and a.order_sk = d.order_sk
and a.order_date_sk = e.date_sk
order by order_number;

        共装载100条销售订单数据，最后20条如图1所示。


图1