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 条评论
登录 后参与评论

相关文章

来自专栏Java后端技术

SpringBoot各类扩展点详解

  上篇文章我们深入分析了SpringBoot的一站式启动流程。然后我们知道SpringBoot的主要功能都是依靠它内部很多的扩展点来完成的,那毋容置疑,这些扩...

903
来自专栏流柯技术学院

Monkey测试2——Monkey测试策略

Monkey的测试策略 一. 分类 Monkey测试针对不同的对象和不同的目的采用不同的测试方案,首先测试的对象、目的及类型如下: 测试的类型分为:应用...

703
来自专栏企鹅号快讯

分布式系统一致性分类,你知道几种?

为了提升系统的可用性、性能、扩展性,我们可以从两个方面着手, 要去建立多个副本。可以放到不同的物理机、机架、机房、地域。一个副本的失效可以让请求转到其他副本。 ...

1.1K10
来自专栏架构师之路

细聊分布式ID生成方法

一、需求缘起 几乎所有的业务系统,都有生成一个记录标识的需求,例如: (1)消息标识:message-id (2)订单标识:order-id (3)帖子标识:t...

3415
来自专栏跟着阿笨一起玩NET

.NET轻量级DBHelpers数据访问组件

553
来自专栏XAI

SpringMVC+MongoDB+Maven整合(微信回调Oauth授权)

个人小程序。里面是基于百度大脑 腾讯优图做的人脸检测。是关于人工智能的哦。 2017年第一篇自己在工作中的总结文档。土豪可以打赏哦。 https://git.o...

5877
来自专栏进击的程序猿

高效的并发控制

本文是阅读论文Efficient Optimistic Concurrency Control Using Loosely Synchronized Clock...

603
来自专栏用户画像

3.3 分页管理与分段管理的比较

分页 管理方式和分段管理方式在很多地方相似,比如内存中都是不连续的,都有地址变换机构来进行地址映射等。但两者也存在着很多区别。

642
来自专栏数据和云

如何提高Linux下块设备IO的整体性能?

编辑手记:本文主要讲解Linux IO调度层的三种模式:cfp、deadline和noop,并给出各自的优化和适用场景建议。 作者简介: ? 邹立巍 Linux...

4574
来自专栏加米谷大数据

Spark的性能调优

下面这些关于Spark的性能调优项,有的是来自官方的,有的是来自别的的工程师,有的则是我自己总结的。

1042

扫码关注云+社区