Sqoop导入数据到Hadoop代理执行，sqoophadoop

Sqoop导入数据到Hadoop代理执行，sqoophadoop

        最近在做执行服务器，它根据用户输入的sqoop命令代理向hadoop提交任务执行，目前需要支持的数据源包括mysql、oracle以及公司自己的分布式数据库DDB，数据导入的目的地可以是HDFS或者hive表。         首先来讨论一下对hive的支持，hive是作为一个支持JDBC的数据库，它的数据分成两部分，元数据和数据，元数据保存在一个本地的数据库，例如嵌入式数据库derby或者mysql，主要是存储一些关于hive的数据库和表定义的一些信息（关于元数据库表需要补充一下，这些表的创建都是hive完成的，不需要我们预先创建，但是在mysql中需要将字符集设置为latin1，否则在hive操作的时候可能出现问题，参见http://www.cnblogs.com/blueren/archive/2011/06/29/sir_001.html），数据保存在hadoop中，它支持将本地文件导入到hive中，其实hive在导入数据库的时候并不会解析数据库，而是是将文件存储在hadoop中，对于数据的来源，它只支持load这样的全量导入和批量导入的操作，并不支持一条条的insert操作和update操作，对hive的查询是利用hadoop的计算能力将select语句转换成一个或者多个hadoop任务，通过hadoop的计算得到SQL的执行结果。所以对于sqoop来说，导入到hive其实也是将数据导入到HDFS中，只不过hive需要对元数据进行操作，在sqoop中，导入数据到hive执行的操作和导入数据到HDFS大体上流程是差不多的，只不过导入到HDFS中是向hadoop提交一个job执行数据导入到HDFS的某个文件，而导入到hive的时候在将数据放到HDFS之后在生成一个hive的脚本，然后调用hive程序（或者直接调用命令行中的hive命令）使用"-f"参数来执行该文件，例如我将一个mysql中的表pf_sqoop导入到hive中的demo_blog_4表中，当前的hadoop用户是hive用户，该用户是代理intern用户执行的操作，生成的hive的脚本文件内容如下：

CREATE TABLE `demo_blog_4` ( `jobid` BIGINT, `product` STRING, `email` STRING, `jobname` STRING, `jobtype` TINYINT, `stype` TINYINT, `dtype` TINYINT, `dir` STRING, `dburl` STRING, `tablename` STRING, `username` STRING, `password` STRING, `hdbname` STRING, `htable` STRING, `params` STRING, `cluster` STRING, `begin_time` BIGINT, `end_time` BIGINT, `duration` BIGINT, `status` TINYINT, `created_at` STRING, `updated_at` STRING) COMMENT 'Imported by sqoop on 2015/04/13 10:16:31' ROW FORMAT DELIMITED FIELDS TERMINATED BY '\001' LINES TERMINATED BY '\012' STORED AS TEXTFILE;
LOAD DATA INPATH 'hdfs://hp1/user/intern/pf_sqoop' OVERWRITE INTO TABLE `demo_blog_4`;<span style="font-family: Tahoma; background-color: rgb(255, 255, 255);">     </span>

        可以看出，我们的数据库表的数据已经导入到HDFS的hdfs://hp1/user/intern/pf_sqoop路径下（这应该是/user/user_name/表名），然后执行创建hive表，再将该文件的数据导入到hive中，这里导入到hive表中是将已经存在hdfs中的文件再导入到hive的数据目录下。sqoop关于hive的参数有以下几个： --create-hive-table                  //是否创建新表，覆盖原来的表定义 --hive-overwrite                     //是否覆盖hive中已经存在的数据 --hive-import                         //标记是否导入数据到hive --hive-home                          //hive的主目录，必要的时候需要根据该配置找到hive命令 --hive-table                           //执行hive的表名
        接下来看一下导入数据到HDFS吧，首先我们使用sqoop来执行mysql和HDFS之间的导入和导出，sqoop对于mysql的支持还是比较好的，实验一下就能够成功了，使用的命令如下： sqoop import --table pf_sqoop --target-dir /user/intern/terry  --delete-target-dir  -m 1 --connect jdbc:mysql://ip:port/database?user=name\&password=passwd --input-fields-terminated-by ','，执行成功之后该目录/user/intern/terry下会创建导入的表数据的文件。--delete-target-dir参数意味着在生成数据之前将目标路径清空，也就意味着会覆盖之前的数据。
然后再用oracle，开始的时候我想执行oracle的driver吧（因为sqoop使用的默认的driver和我之前使用的不一样），于是添加了这样的参数--driver oracle.jdbc.driver.OracleDriver，但是结果出现了如下的错误：

Beginning code generation
Executing SQL statement: SELECT t.* FROM test AS t WHERE 1=0
Error executing statement: java.sql.SQLException: ORA-00933: SQL command not properly ended

java.sql.SQLException: ORA-00933: SQL command not properly ended 

        根据错误堆栈查看代码发现执行的操作是sqoop需要查看数据源中该表的表结构，然后根据该结果生成java代码，从日志中可以看到查看表结果的SQL是SELECT t.* FROM test AS t WHERE 1=0，但是使用jdbc对oracle进行测试发现这个SQL是不支持的（不支持表的as），查看代码发现在执行importTable的时候执行了如下的操作：

jarFile = codeGenerator.generateORM(options, tableName);

        该操作是根据表名生成jar文件，在该函数中调用getColumnTypes函数生成表的字段名和每一个字段的数据类型，该函数中执行如下：

1831  protected Map<String, Integer> getColumnTypes() throws IOException {
1832    if (options.getCall() == null) {
1833      return connManager.getColumnTypes(tableName, options.getSqlQuery());
1834    } else {
1835      return connManager.getColumnTypesForProcedure(options.getCall());
1836    }
1837  }

        可以看出，真正执行的操作是在connManager中完成的，那么这个connManager是什么呢？发现在sqoop中有如下一个参数：--connection-manager  className这里指定的是连接管理类，开始以为是dbcp之类的呢，但是我错了，这里指定的是在sqoop内部的数据库操作类，如果不指定这个参数会使用哪个connManager呢？代码如下：

121  public ConnManager getManager(JobData data) throws IOException {
122    com.cloudera.sqoop.SqoopOptions options = data.getSqoopOptions();
123    String manualDriver = options.getDriverClassName();
124    String managerClassName = options.getConnManagerClassName();
125   <span style="color:#ff6666;"> //如果在命令中指定了driver但是没有指定connManager那么就使用通用的JDBC的connManager，也会输出warn日志</span>
131    if (manualDriver != null && managerClassName == null) {
132      LOG.warn("Parameter --driver is set to an explicit driver however"
133        + " appropriate connection manager is not being set (via"
134        + " --connection-manager). Sqoop is going to fall back to "
135        + GenericJdbcManager.class.getCanonicalName() + ". Please specify"
136        + " explicitly which connection manager should be used next time."
137      );
138      return new GenericJdbcManager(manualDriver, options);
139    }
140   <span style="color:#ff6666;"> //如果制定了connManager，直接创建该类对象，不存在则报错。</span>
142    if (managerClassName != null){
143      ConnManager connManager = null;
144
145      try {
146        Class<ConnManager> cls = (Class<ConnManager>)
147          Class.forName(managerClassName);
148
152        if (manualDriver == null) {
153          Constructor<ConnManager> constructor =
154            cls.getDeclaredConstructor(com.cloudera.sqoop.SqoopOptions.class);
155          connManager = constructor.newInstance(options);
156        } else {
157          Constructor<ConnManager> constructor =
158            cls.getDeclaredConstructor(String.class,
159                                       com.cloudera.sqoop.SqoopOptions.class);
160          connManager = constructor.newInstance(manualDriver, options);
161        }
162      } catch (ClassNotFoundException e) {
163        LOG.error("Sqoop could not found specified connection manager class "
164          + managerClassName  + ". Please check that you