基于Hadoop生态圈的数据仓库实践 —— ETL（一）

    Sqoop是一个在Hadoop与结构化数据存储（如关系数据库）之间高效传输大批量数据的工具。它在2012年3月被成功孵化，现在已是Apache的顶级项目。Sqoop有Sqoop1和Sqoop2两代，Sqoop1最后的稳定版本是1.4.6，Sqoop2最后版本是1.99.6。需要注意的是，1.99.6与1.4.6并不兼容，而且截止目前为止，1.99.6并不完善，不推荐在生产环境中部署。

    Sqoop1的架构图如下所示。

    第一代Sqoop的设计目标很简单：

    Sqoop1是用Java开发的，完全客户端驱动，严重依赖于JDBC，可以使用简单的命令行命令导入导出数据。例如：

    # 导入

sqoop import --connect jdbc:mysql://localhost/testdb --table PERSON --username test --password ****

    上面这条命令形成一系列任务：

    # 导出

sqoop export --connect jdbc:mysql://localhost/testdb --table CLIENTS_INTG --username test --password **** --export-dir /user/localadmin/CLIENTS

    上面这条命令形成一系列任务：

    Sqoop1有许多简单易用的特性，如可以在命令行指定直接导入至Hive、HDFS或HBase。连接器可以连接大部分流行的数据库：Oracle、SQLServer、MySQL、Teradata、PostgreSQL等。

    Sqoop1的主要问题包括：

    Sqoop2的架构图如下所示。

    Sqoop2体系结构比Sqoop1复杂得多，被设计用来解决Sqoop1的问题。

    易用性

    Sqoop1需要客户端的安装和配置，而Sqoop2是在服务器端安装和配置。这意味着连接器只在一个地方统一配置，由管理员角色管理，操作员角色使用。类似地，只需要在一台服务器上配置JDBC驱动和数据库连接。Sqoop2还有一个基于Web的服务：前端是命令行接口（CLI）和浏览器，后端是一个元数据知识库。用户可以通过Web接口进行导入导出，避免的错误选项和繁冗的步骤。此外，Sqoop2还在服务器端整合了Hive和HBase。Oozie通过REST API管理Sqoop任务，这样当安装一个新的Sqoop连接器后，无需在Oozie中安装它。

    可扩展性

    在Sqoop2中，连接器不再受限于JDBC词汇（必须指定database、table等），它甚至可以定义自己使用的词汇。例如，Couchbase不需要指定表名，只需在充填或卸载操作时重载它。通用的功能将从连接器中抽取出来，使之只负责数据传输。在Reduce阶段实现通用功能，确保连接器可以从将来的功能性开发中受益。连接器不再需要提供与其它系统整合等下游功能，因此，连接器的开发者不再需要了解所有Sqoop支持的特性。

    安全性

    当前，用户是通过执行‘sqoop’命令运行Sqoop。Sqoop作业的安全性主要由对执行Sqoop的用户信任所决定。Sqoop2将作为基于应用的服务，通过按不同角色连接对象，支持对外部系统的安全访问。为了进一步安全，Sqoop2不再允许生成代码、请求直接访问Hive或HBase，也不对运行的作业开放访问所有客户端的权限。Sqoop2将连接作为一级对象。包含证书的连接一旦生成，可以被不同的导入导出作业多次使用。连接由管理员生成，被操作员使用，因此避免了最终用户的权限泛滥。此外，连接还可以被限制只能进行某些基本操作（如导入导出）。通过限制同一时间打开连接的总数和一个禁止连接的选项来管理资源。

    CDH 5.7.0中的Sqoop既包含Sqoop1又包含Sqoop2，Sqoop1的版本是1.4.6，Sqoop2的版本是1.99.5。当前的Sqoop2还缺少Sqoop1的某些特性，因此Cloudera的建议是，只有当Sqoop2完全满足需要的特性时才使用它，否则继续使用Sqoop1。CDH 5.7.0中的Sqoop1和Sqoop2的特性区别如下表所示。

    在本示例中使用Sqoop1从MySQL库抽取数据到Hive。

    从源抽取数据导入数据仓库（本示例的RDS）有两种方式，可以从源把数据抓取出来（拉），也可以请求源把数据发送（推）到数据仓库。影响选择数据抽取方式的一个重要因素是操作型系统的可用性和数据量，这基于是抽取整个数据还是仅仅抽取自最后一次抽取以来的变化数据。考虑以下两个问题：

    完全抽取和变化数据捕获（CDC） 

    如果数据量很小并且易处理，一般来说采取完全源数据抽取（将所有的文件记录或所有的数据库表数据抽取至数据仓库）。这种方式适合引用类型的源数据，比如邮政编码。引用型源数据通常是维度表的源。如果源数据量很大，抽取全部数据是不可行的，那么只能抽取变化的源数据（自最后一次抽取以来变化的数据）。这种数据抽取模式称为变化数据捕获（CDC），通常被用于抽取操作型系统的事务数据，比如销售订单。

    CDC大体可以分为两种，一种是侵入式的，另一种是非侵入式的。所谓侵入式的是指CDC操作会给源系统带来性能的影响。只要CDC操作以任何一种方式执行了SQL语句，就可以认为是侵入式的CDC。常用的四种CDC方法中有三种是侵入性的，这四种方法是：基于时间戳的CDC、基于触发器的CDC、基于快照的CDC、基于日志的CDC。下表总结了四种CDC方案的特点。

    从源拉数据或源来推数据

    如果想让数据源只是简单的等待数据仓库来抽取，那么可以使用拉模式。但是必须确认，在数据仓库抽取数据时，源数据必须是可用的而且已经准备好了数据。如果抽取数据的实时性非常重要，或者希望数据源一旦准备好数据就立即发送，那么应该使用由数据源推数据的抽取模式。如果数据源是受到保护并且是禁止访问的，则只能使用数据源推数据的方式。

    下表中汇总了本示例中维度表和事实表用到的源数据表及其抽取模式。

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

sqoop import --connect jdbc:mysql://cdh1:3306/source?useSSL=false --username root --password mypassword --table customer --hive-import --hive-table rds.customer --hive-overwrite
sqoop import --connect jdbc:mysql://cdh1:3306/source?useSSL=false --username root --password mypassword --table product --hive-import --hive-table rds.product --hive-overwrite

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

    Sqoop支持两种类型的增量导入：append和lastmodified。可以使用--incremental参数指定增量导入的类型。

    当被导入表的新行具有连续递增的行id值时，应该使用append模式。指定行id为--check-column的列。Sqoop导入那些被检查列的值比--last-value给出的值大的数据行。

    Sqoop支持的另一个表修改策略叫做lastmodified模式。当源表的数据行可能被修改，并且每次修改都会更新一个last-modified列为当前时间戳时，应该使用lastmodified模式。那些被检查列的时间戳比--last-value给出的时间戳新的数据行被导入。

    在增量导入的最后，后续导入使用的--last-value会被打印出来。当执行后面的导入时，应该用这种方式指定--last-value参数的值，以确保只导入新的或修改过的数据。可以通过一个增量导入的保存作业自动执行这个过程，这是适合重复执行增量导入的方式。

    下面测试一下增量导入：

sqoop job --create myjob_1 \
-- \
import \
--connect "jdbc:mysql://cdh1:3306/source?useSSL=false&user=root&password=mypassword" \
--table sales_order \
--columns "order_number, customer_number, product_code, order_date, entry_date, order_amount" \
--where "entry_date < current_date()" \
--hive-import \
--hive-table rds.sales_order \
--incremental append \
--check-column entry_date \
--last-value '1900-01-01'

    说明：在作业中使用的--where参数，是为了只导入前一天的数据。

sqoop job --show myjob_1 | grep last.value

    可以看到，last-value的值为初始的'1900-01-01'

sqoop job --exec myjob_1

sqoop job --show myjob_1 | grep last.value

    可以看到，last-value的值为当前最大值'2016-06-30 05:20:47'

SET @customer_number := floor(1 + rand() * 6);  
SET @product_code := floor(1 + rand() * 2);  
SET @order_date := from_unixtime(unix_timestamp('2016-07-03') + rand() * (unix_timestamp('2016-07-04') - unix_timestamp('2016-07-03')));
SET @amount := floor(1000 + rand() * 9000);

INSERT INTO sales_order 
VALUES (101,@customer_number,@product_code,@order_date,@order_date,@amount);

SET @customer_number := floor(1 + rand() * 6);
SET @product_code := floor(1 + rand() * 2);
SET @order_date := from_unixtime(unix_timestamp('2016-07-04') + rand() * (unix_timestamp('2016-07-05') - unix_timestamp('2016-07-04')));
SET @amount := floor(1000 + rand() * 9000);

INSERT INTO sales_order 
VALUES (102,@customer_number,@product_code,@order_date,@order_date,@amount);

COMMIT;

    上面的语句向sales\_order插入了两条记录，一条是7月3日的，另一条是7月4日的，如下图所示。

sqoop job --exec myjob_1

sqoop job --show myjob_1 | grep last.value

    可以看到，last-value的值已经变为'2016-07-03 22:45:46'

select * from sales_order order by order_number desc;

    结果如下图所示，可以rds.sales\_order表中只新增了一条数据，7月4日的记录被作业中的where过滤掉。

    至此介绍了使用Sqoop抽取数据的一般方法。Sqoop有众多的命令行参数，具体可参考官方文档，链接地址如下：[http://sqoop.apache.org/docs/1.4.6/SqoopUserGuide.html](http://sqoop.apache.org/docs/1.4.6/SqoopUserGuide.html)