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

标签：

        Sqoop1的架构图如下所示。

• 在企业级数据仓库、关系数据库、文档系统和Hive、HDFS之间导入导出数据。
• 基于客户端的模型。
• 连接器使用厂商提供的驱动。
• 没有集中的元数据存储。
• 只有Map任务，没有Reduce任务。数据传输和转化都由Mappers提供。
• 可以使用Oozie调度和管理Sqoop作业。

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

• 生成SQL代码。
• 执行SQL代码。
• 生成Map作业。
• 执行Map作业。
• 数据传输到HDFS。

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

• 生成Map作业。
• 执行Map作业。
• 从HDFS的/user/localadmin/CLIENTS路径传输数据。
• 生成SQL代码。
• 向CLIENTS_INTG插入数据。

• 繁多的命令行参数。
• 不安全的连接方式（直接在命令行写密码等）。
• 没有元数据存储，只能本地配置和管理，使复用受限。

        Sqoop2的架构图如下所示。

• 需要抽取哪部分源数据加载到数据仓库？有两种方式，完全抽取和变化数据捕获。
• 数据抽取的方向是什么？有两种方式，拉模式（从数据仓库去拉）和推模式（通过源去推）。

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

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

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增量导入的基本了解，下面看一下如何在本示例中使用它抽取数据。对于sales_order这个表采用基于时间戳的CDC拉取方式抽数据。这里假设源系统中销售订单记录一旦入库就不再改变，或者可以忽略改变。也就是说销售订单是一个随时间变化单向追加数据的表。sales_order表中有两个关于时间的字段，order_date表示订单时间，entry_date表示订单数据实际插入表里的时间，在后面讨论“迟到的事实”时就会看到两个时间可能不同。那么用哪个字段作为CDC的时间戳呢？设想这样的场景，一个销售订单的订单时间是2015年1月1日，实际插入表里的时间是2015年1月2日，ETL每天0点执行，抽取前一天的数据。如果按order_date抽取数据，条件为where order_date >= ‘2015-01-02‘ AND order_date < ‘2015-01-03‘，则2015年1月3日0点执行的ETL不会捕获到这个新增的订单数据。所以应该以entry_date作为CDC的时间戳。

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

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 >= date_add(current_date(),interval -1 day) and entry_date < current_date()" --hive-import --hive-table rds.sales_order --incremental append --check-column entry_date --last-value '1900-01-01'

insert into source.sales_order values(null,7,3,date_add(current_date(),interval -1 day),date_add(current_date(),interval -1 day),10000);
commit;

sqoop job --exec myjob_1

select * from sales_order order by order_number desc;

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

标签：

原文地址：http://blog.csdn.net/wzy0623/article/details/51804557