小编sri*_*hna的帖子

在我们的组织中,我们一直在尝试使用基于hadoop生态系统的工具来实现ETL.虽然生态系统本身很大,但目前我们只使用一套非常有限的工具.我们的典型管道流程如下:

Source Database (1 or more) -> sqoop import -> pig scripts -> sqoop export -> Destination Database (1 or more)

在一段时间内,我们遇到了上述实施ETL方法的多个问题.我们注意到的一个问题是,当尝试使用pig从HDFS读取字段时,字段无法正确对齐(其中HDFS上的数据通常通过sqoop导入)并且pig脚本因错误而失败.例如,由于未对齐,字符串可能以数字字段类型结束.

看来这个问题有两种方法:

1. 在使用pig处理之前,删除字段中已知的问题字符.这是我们过去采取的方法.我们知道我们的源数据库中有一些不良数据 - 通常是不存在的字段中的新行和制表符.(注意:我们曾经将标签作为字段分隔符).所以我们所做的就是使用DB视图或自由格式查询选项sqoop,后者又使用REPLACE函数或其在源数据库中可用的等效函数(通常是mysql,但不常使用postgres).这种方法确实有效,但它具有HDFS数据与源数据不匹配的副作用.此外,其他一些导入的字段将不再有意义 - 例如,假设您在某个字段上有一个MD5或SHA1哈希,但该字段已被修改以替换某些字符,所以我们必须计算MD5或SHA1是否一致,而不是从源DB中导入一个.此外,这种方法在一定程度上涉及反复试验.我们不一定知道哪些字段需要提前修改(以及要删除的字符),因此我们可能需要多次迭代才能达到最终目标.

2. 使用带有sqoop的机箱功能与转义相结合,并将其与pig中适当类型的加载器结合使用,这样不仅字段可以正确排列,而且给定字段(及其关联值)的表示方式与数据相同穿过管道.

我试图用sqoop和pig中可用的不同选项找出完成#2的好方法.下面给出的是除了调查结果之外我迄今为止所尝试的内容.

以下是用于此实验的软件的特定版本:

Sqoop: 1.4.3
Pig: 0.12.0
Hadoop: 2.0.0

由于我们的数据集通常很大(并且需要几个小时才能处理),我想我会想出一个非常小的数据集,它模仿我们遇到的一些数据问题.为此,我在mysql中放了一个小表(将用作源数据库):

mysql> desc example;
+-------+---------------+------+-----+---------+----------------+
| Field | Type          | Null | Key | Default | Extra          |
+-------+---------------+------+-----+---------+----------------+
| id    | int(11)       | NO   | PRI | NULL    | auto_increment |
| name  | varchar(1024) | YES  |     | NULL    |                | …

我注意到sqoop import的一个奇怪问题.我试图导入的数据在MySQL DB中采用以下形式:

<a1, a2, a3, d1, a4, a5, a6, a7, a8>

其中a1,a2,...,a8是varchar类型,d1是时间戳类型.由于数据的规范化,我不得不做JOIN多个表来获取这些列 - 如下所示:

SELECT t1.a1, t2.a2....... from table t1 INNER JOIN table t2 ON t1.t2_id = t2.id ........... WHERE <some condition>

最初我开始时有这样的观点:

CREATE OR REPLACE VIEW my_view AS
SELECT t1.a1, t2.a2....... from table t1 INNER JOIN table t2 ON t1.t2_id = t2.id ........... WHERE <some condition>

使用此视图导入sqoop命令,如下所示:

sqoop import --connect [jdbc url] --username [user] --password [password] --table my_view --target-dir my_dir --split-by a5 --mysql-delimiters --verbose --boundary-query 'SELECT min(a5), max(a5) …