《Hadoop实战手册》一1.5 使用Sqoop从HDFS导出数据到MySQL

本节书摘来异步社区《Hadoop实战手册》一书中的第1章，第1.5节，作者： 【美】Jonathan R. Owens , Jon Lentz , Brian Femiano 译者： 傅杰 , 赵磊 , 卢学裕 责编： 杨海玲，更多章节内容可以访问云栖社区“异步社区”公众号查看。

1.5 使用Sqoop从HDFS导出数据到MySQL

Sqoop是Apache基金会下的一个项目，是庞大Hadoop生态圈中的一部分。在很多方面Sqoop和distcp很相似（见1.3节）。这两个工具都是构建在MapReduce之上的，利用了MapReduce的并行性和容错性。与集群间的数据复制不同，Sqoop设计通过JDBC驱动连接实现Hadoop集群与关系数据库之间的数据复制。

它的功能非常广泛，本节将以网络日志条目为例展示如何使用Sqoop从HDFS导入数据到MySQL数据库。

准备工作
本例使用Sqoop V1.3.0版本。

如果你使用的是CDH3版本，Sqoop默认是已经安装了。如果不是CDH3，你可以通过https://ccp.cloudera.com/display/CDHDOC/Sqoop+Installation找到发行版的说明。

在本节假设你已经启动了一个MySQL实例，并且能够访问Hadoop集群。mysql.user表配置了你运行Sqoop的那台机器上被允许连接的用户。访问http://dev.mysql.com/doc/refman/5.5/en/installing.html获取更多关于MySQL安装与配置的相关信息。

将MySQL JDBC驱动包复制到$SQOOP_HOME/libs目录下。该驱动包可以从http://dev.mysql.com/downloads/connector/j/ 下载。

按照1.1节介绍的导入weblog_entires.txt文件到HDFS的方式操作。

操作步骤
完成以下步骤实现将HDFS数据导出到MySQL表中。

1．在MySQL实例中创建一个新数据库：

CREATE DATABASE logs;

        

          
        
        
        

          

          AI 代码解读
        
      
      

2．创建表weblogs_from_hdfs：

USE logs;
CREATE TABLE weblogs_from_hdfs (
    md5             VARCHAR(32),
    url             VARCHAR(64),
    request_date    DATE,
    request_time    TIME,
    ip              VARCHAR(15)
);

        

          
        
        
        

          

          AI 代码解读
        
      
      

3．从HDFS导出weblog_entries.txt文件到MySQL：

sqoop export -m 1 --connect jdbc:mysql://<HOST>:<PORT>/logs --username hdp_usr 
--password test1 --table weblogs_from_hdfs --export-dir /data/weblogs/05102012 
--input-fields-terminated-by '\t' --mysql-delmiters

        

          
        
        
        

          

          AI 代码解读
        
      
      

输出结果如下：

INFO mapreduce.ExportJobBase: Beginning export of weblogs_from_
hdfs
input.FileInputFormat: Total input paths to process : 1
input.FileInputFormat: Total input paths to process : 1
mapred.JobClient: Running job: job_201206222224_9010
INFO mapred.JobClient: Map-Reduce Framework
INFO mapred.JobClient: Map input records=3000
INFO mapred.JobClient: Spilled Records=0
INFO mapred.JobClient: Total committed heap usage
(bytes)=85000192
INFO mapred.JobClient: Map output records=3000
INFO mapred.JobClient: SPLIT_RAW_BYTES=133
INFO mapreduce.ExportJobBase: Transferred 248.3086 KB in 12.2398
seconds (20.287 KB/sec)
INFO mapreduce.ExportJobBase: Exported 3000 records.

        

          
        
        
        

          

          AI 代码解读
        
      
      

工作原理
Sqoop连接数据库的JDBC驱动可使用-connect参数声明，并从$SQOOP_HOME/libs目录中加载相应的包。其中$SQOOP_HOME为Sqoop安装的绝对路径。--username和--password选项用于验证用户访问MySQL实例的权限。mysql.user表必须包含Hadoop集群每个节点的主机域名以及相应的用户名，否则Sqoop将会抛出异常，表明相应的主机不允许被连接到MySQL服务器。

mysql> USE mysql; 
mysql> select host, user from user; 
+-----------------+------------+ 
| user            | host       | 
+-----------------+------------+ 
| hdp_usr         | hdp01      | 
| hdp_usr         | hdp02      | 
| hdp_usr         | hdp03      | 
| hdp_usr         | hdp04      | 
| root            | 127.0.0.1  | 
| root            | ::1        | 
| root            | localhost  | 
+-----------------+------------+ 
7 rows in set (1.04 sec)

        

          
        
        
        

          

          AI 代码解读
        
      
      

在这个例子中，我们使用hdp_usr用户连接到MySQL服务器。我们的集群拥有4台机器，即hdp01、hdp02、hdp03和hdp04。

--table参数决定了HDFS导出的数据将存储在哪个MySQL表中。这个表必须在执行Sqoop export语句之前创建好。Sqoop通过表的元数据信息、列数量以及列类型来校验HDFS需要导出目录中的数据并生成相应的插入语句。举个例子，导出作业可以被想象为逐行读取HDFS的weblogs_entries.txt文件并产生以下输出：

INSERT INTO weblogs_from_hdfs 
VALUES('aabba15edcd0c8042a14bf216c5', '/jcwbtvnkkujo.html', '2012-05- 10',   
'21:25:44', '148.113.13.214'); 

INSERT INTO weblogs_from_hdfs 
VALUES('e7d3f242f111c1b522137481d8508ab7', '/ckyhatbpxu.html', '2012- 05-10',   
'21:11:20', '4.175.198.160');

INSERT INTO weblogs_from_hdfs 
VALUES('b8bd62a5c4ede37b9e77893e043fc1', '/rr.html', '2012-05-10', '21:32:08',   
'24.146.153.181'); 
...

        

          
        
        
        

          

          AI 代码解读
        
      
      

Sqoop export默认情况下是创建新增语句。如果--update-key参数被设置了，则将是创建更新语句。如果前面的例子使用了参数--update-key md5那么生成的Sql代码将运行如下：

UPDATE weblogs_from_hdfs SET url='/jcwbtvnkkujo.html', request_ date='2012-   
05-10'request_time='21:25:44' 
ip='148.113.13.214'WHERE md5='aabba15edcd0c8042a14bf216c5' 

UPDATE weblogs_from_hdfs SET url='/jcwbtvnkkujo.html', request_ date='2012-05-   
10' request_time='21:11:20' ip='4.175.198.160' WHERE md5='e7d3f242f111c1b   
522137481d8508ab7' 

UPDATE weblogs_from_hdfs SET url='/jcwbtvnkkujo.html', request_ date='2012-   
05-10'request_time='21:32:08' ip='24.146.153.181' WHERE md5='b8bd62a5c4ede37b   
9e77893e043fc1'

        

          
        
        
        

          

          AI 代码解读
        
      
      

如果--update-key设置的值并没找到，可以设置--update-mode为allowinsert允许新增这行数据。

-m参数决定将配置几个mapper来读取HDFS上文件块。每个mapper各自建立与MySQL服务器的连接。每个语句将会插入100条记录。当完成100条语句也就是插入10000条记录，将会提交当前事务。一个失败的map任务，很可能导致数据的不一致，从而出现插入冲突数据或者插入重复数据。这种情况可以通过使用参数--staging-table来解决。这会促使任务将数据插入一个临时表，等待一个事务完成再将数据从临时表复制到--table参数配置的表中。临时表结构必须与最终表一致。临时表必须是一个空表否则需要配置参数--clear-staging-table。

延伸阅读

• 使用Sqoop从MySQL数据库导入HDFS（1.4节）。