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

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

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

源数据表

HDFS目录

对应EXT模式中的表

抽取模式

customer

/data/ext/customer

customer

整体、拉取

product

/data/ext/product

product

整体、拉取

sales_order

/data/ext/sales_order

sales_order

基于时间戳的CDC、拉取

表1

1. 覆盖导入 

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

2. 增量导入

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

参数

描述

--check-column

在确定应该导入哪些行时,指定被检查的列。列不能是CHAR/NCHAR/VARCHAR/VARNCHAR/LONGVARCHAR/LONGNVARCHAR数据类型。

--incremental

指定Sqoop怎样确定哪些行是新行。有效值是append和lastmodified。

--last-value

指定已经导入数据的被检查列的最大值。

表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表相对应,而销售订单事务表是多个数据仓库表的数据源。

源数据

源数据类型

文件名/表名

数据仓库中的目标表

客户

MySQL表

customer

customer_dim

产品

MySQL表

product

product_dim

销售订单

MySQL表

sales_order

order_dim、sales_order_fact

表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

本文参与腾讯云自媒体分享计划,欢迎正在阅读的你也加入,一起分享。

发表于

我来说两句

0 条评论
登录 后参与评论

相关文章

来自专栏Hadoop数据仓库

HAWQ取代传统数仓实践(四)——定期ETL(Sqoop、HAWQ)

一、变化数据捕获(CDC)         初始装载只在数据仓库开始使用前执行一次,然而,必须要周期性地执行装载源数据过程。与初始装载不同,定期装载一般都是增量...

44710
来自专栏极客编程

测试利器Mocha

mocha 是一个功能丰富的javascript测试框架,可以运行在nodejs和浏览器环境,使异步测试变得简单有趣。mocha 串联运行测试,允许灵活和精确地...

632
来自专栏Golang语言社区

Go1.7改善了编译速度并且会生成更快的代码

Go1.7的开发周期正在接近它的下一个里程碑,Go的提交者Dave Cheney报告了子即将发布的版本中,团队成员在语言工具链上的努力。 Cheney称,基于当...

3359
来自专栏Java学习网

Java初学人员的注意事项

Java中J2SE J2EE J2ME的区别 多数编程语言都有预选编译好的类库以支持各种特定的功能,在Java中,类库以包(package)的形式提...

2188
来自专栏腾讯云移动服务团队

小程序·云开发的云函数路由高级玩法

在掘金开发者大会上,在推荐实践那里,我有提到一种云函数的用法,我们可以将相同的一些操作。

3715
来自专栏码云1024

托管C++、C++/CLI、CLR

2754
来自专栏架构之路

Hive简介

转载自http://www.cnblogs.com/sharpxiajun/archive/2013/06/02/3114180.html Hive简介   首...

2643
来自专栏恰同学骚年

.NET单元测试的艺术-3.测试代码

开篇:上一篇我们学习单元测试和核心技术:存根、模拟对象和隔离框架,它们是我们进行高质量单元测试的技术基础。本篇会集中在管理和组织单元测试的技术,以及如何确保在真...

643
来自专栏安智客

GP TEE标准体系一张图

昨天提到的TEE测试认证,我们知道TEE的功能性测试也就是按照GP TEE API规范的测试,那么我们今天来看看GP TEE的规范有那些。 ...

2065
来自专栏ThoughtWorks

Rec:一个项目的诞生|洞见

Rec是一个用来验证和转换数据文件的Java应用。从第一行代码到v1版本成形,仅仅经历了一个半月的时间,作为一个开源项目,在很多方面都有着各种各样的纠结。 ? ...

2894

扫码关注云+社区