Sqoop的Mysql数据导出实现分两种，一种是使用JDBC方式从Mysql中获取数据，一种是使用MysqlDump命令从MySql中获取数据，默认是 JDBC方式获取数据，如果要使用dump方式获取数据，需要添加 -direct 参数。

使用JDBC方式从Mysql中获取数据

配置语句时，需要添加 $CONDITIONS 点位符，比如：SELECT id FROM user WHERE $CONDITIONS，Sqoop在内部实现时会把它替换成需要的查询条件。

Sqoop启动后会先查询元数据，它会把 $CONDITIONS 替换为 (1=0) ，然后用得到的SQL语句查询数据表对应的Meta信息对于导出一个表的情况，Sqoop会使用这个SQL查询三次数据库，分别是： 1、获取 colInfo（最终得到columnTypes信息）2、查询ColumnNames信息3、生成QueryResult类执行 generateFields操作获取columnTypeNames时。

Sqoop会对获取的Fields做校验，列不能重复，它还会处理数据库的字段到Java属性名的转换

QueryResult类是通过构建java类文件，然后获取JavaCompiler，然后编译加载，为了提高处理性能，不是使用反射实现的，这个生成类内部处理mysql到hdfs属性值为空和分隔符的处理。

接着它会进行下面一个Sql查询操作，查询结果集为MIN(split列),MAX(split列)，查询条件的处理逻辑为 $CONDITIONS 替换为(1=1)，然后再添加外面SELECT查询 (举例：SELECT MIN(id), MAX(id) FROM (SELECT ID,NAME,PASSPORT WHERE (1=1) ) AS t1 )，这样就查询出来此次导出数据最大的split列值和最小的split列值。

对于为整数、布尔值、时间格式、Float等 的分区列，进行split时直接根据对应值的大小进行Split，Text文本的处理方式比较特殊，Sqoop先会对之前获取到的Min和Max的字串寻找它们最大的相同前缀子字串，然后把后面的字段转化为BigDecimal，结合时char占两个字节（65536），算法在 TextSplitter类中，比较简单，本质上就是一个进制转换。拆分好后，需要把Split的值再转换为String，然后加上相同前缀子字段，就构成了查询区间了（注意中文可能会被拆分）。

Sqoop对数据的获取是在DataDrivenDBRecordReader中，在查询时会把 $CONDITIONS 替换成 split 的范围比如 ( id >= 1) && (id<10)，使用JDBC获取到结果集游标，然后移动游标处理数据。

使用MysqlDump命令从MySql中获取数据

第二种方法与第一种方式有下面的差别：

初始化元数据，它是在构建的查询语句后面添加 limit 1 ，比如：SELECT t. FROM AS t LIMIT 1，因为dump方式在查询指定获取列时使用的是 t.，当使用limit 0时，数据库不会给它返回必须的元数据信息。

dump方式在map进行数据的获取，其会构建mysqldump命令，然后使用java程序调用，获取输入输出流和错误流，其实现了 org.apache.sqoop.util.AsyncSink 抽象类，用来处理输入输出流和错误流。

优化策略：

Sqoop查询无数据会进行三次相同的Sql查询，可以合并查询，不过由于仅返回Meta信息，查询很快，不需要修改实现。

分区列选择对于查询元数据和导出的查询影响很大，应该对索引做调优，避免对分区列的排序操作，加快元数据查询速度和导出数据的速度，尽量选择自增加的主键ID做Split列，区分度好并且可以顺序读取数据。

导出操作的查询语句中，$CONDITIONS 会被替换为范围区间，创建索引时，要考虑做这个查询的优化。

索引建议，考虑三个规则（使查询数据集较少、减少点的查询、避免排序操作），Sqoop场景下，如果分区列不是主键（自增加）时，把分区列做为联合索引的第一个字段，其它被选择的查询条件做为索引的其它字段。

分区列的选择，要避免Split后数据不均衡。

从实现上来看-m参数是可以增加任务的并行度的，但数据库的读线程是一定的，所以 -m 过大对于数据库会是一个压力，在Sqoop的场景下，数据库是一个影响并发的瓶颈。增加job数意义不大。

下面列出Sqoop目前1.4.6版本存在的两个问题。

查看Sqoop源码，发现其存在两个比较严重的问题。

问题 1、数据分片与Mapper设定不合理

Sqoop在抽取时可以指定 -m 的参数，但这个 -m 的参数是控制mapper的数量的，但它也决定了最后能够生成的文件的数目，调节这个值可以实现对结果文件大小的控制，但问题是，如果产生的文件的格式不能够被分割，那么对这个数据的下游性能有很大影响，同时Sqoop在启动时会启动 -m 个 MapperTask，会对数据库产生m的并发读取。个人认为Sqoop做为一个数据导出框架，对数据的控制应该再细至一些，-m 只是控制MR的并行度，而数据分片数目应该由另外的参数控制。

个人建议可以加个 -split-per-map 参数，比如设置-m=4 -split-per-map=2,则对结果集分 8 片，每个Mapper处理两片数据，最后共产生 8 个文件。

问题 2、分片效率低

Sqoop在做分片处理时有问题，其实现会使用 “Select Max(splitKey),Min(splitKey) From ( –select参数 ) as t1” 的语句来查询分片信息，在Mysql下，这样的查询会产生一个以split-id为主键，包含需要导出的其它所有字段的临时表，如果数据量不大，临时表数据可以在内存中，处理速度还可以保证。但如果数据量很大，内存中已经存放不下时，这些数据会被保存为MyISAM表存放到磁盘文件中，如果数据量再大一些，磁盘文件已经存放不下临时表时，拆分数据会失败。如果数据量很大，这个查询可以占到整个导出时间的45%，优化空间很大，如果不修改实现的话，不适合做大数据量表的全量数据导出操作。

解决方案一：

修改所有导出脚本，分片语句自定义

解决方案二：

修改：org.apache.sqoop.mapreduce.DataDrivenImportJob的

1
2

@Contract(“null, _ -> !null”)
private String buildBoundaryQuery(String col, String query)

修改代码如下

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63

org.apache.sqoop.mapreduce.DataDrivenImportJob的
 
@Contract("null, _ -> !null") 
private String buildBoundaryQuery(String col, String query)
 
/**
   * Build the boundary query for the column of the result set created by
   * the given query.
   * @param col column name whose boundaries we're interested in.
   * @param query sub-query used to create the result set.
   * @return input boundary query as a string
   */
  private String buildBoundaryQuery(String col, String query) {
    if (col == null || options.getNumMappers() == 1) {
      return "";
    }
 
    // Replace table name with alias 't1' if column name is a fully
    // qualified name.  This is needed because "tableName"."columnName"
    // in the input boundary query causes a SQL syntax error in most dbs
    // including Oracle and MySQL.
    String alias = "t1";
    int dot = col.lastIndexOf('.');
    String qualifiedName = (dot == -1) ? col : alias + col.substring(dot);
 
    ConnManager mgr = getContext().getConnManager();
    String ret = mgr.getInputBoundsQuery(qualifiedName, query);
    if (ret != null) {
      return ret;
    }
 
//    return "SELECT MIN(" + qualifiedName + "), MAX(" + qualifiedName + ") "
//        + "FROM (" + query + ") AS " + alias;
    return initBoundaryQuery(qualifiedName, query, alias);
  }
 
  private String initBoundaryQuery(String qualifiedName, String query, String alias) {
    StringBuilder regex = new StringBuilder();
    regex.append("(\\s[A|a][S|s][\\s][`]?");
    for (char c : qualifiedName.toCharArray()) {
      regex.append('[').append(c).append(']');
    }
    regex.append("[`|\\s|,])");
    final Matcher matcher1 = Pattern.compile(regex.toString()).matcher(query);
    final boolean asCheckOk = !matcher1.find();
    if(asCheckOk) {
      final Matcher matcher2 = Pattern.compile("(\\s[F|f][R|r][O|o][M|m]\\s)").matcher(query);
      int count = 0;
      while (matcher2.find()) {
        count++;
      }
      boolean fromCheckOk = count == 1;
      if(fromCheckOk) {
        final Matcher matcher = Pattern.compile("(\\s[F|f][R|r][O|o][M|m]\\s[\\s\\S]*)").matcher(query);
        while (matcher.find()) {
          return "SELECT MIN(" + qualifiedName + "), MAX(" + qualifiedName + ") "
                  + matcher.group();
        }
      }
    }
    return "SELECT MIN(" + qualifiedName + "), MAX(" + qualifiedName + ") "
            + "FROM (" + query + ") AS " + alias;
  }

问题三：无法自定义特殊字符替换

解决方案一：

通过SQL的替换功能，修改脚本代价高，并且脚本可读性、可维护性都大大降低

解决文案二：

修改Sqoop实现，增加自定义特殊字符替换功能

协议：

输出文件压缩：Camus默认只支持两种压缩格式(snappy和deflate)，默认是defalte，使用 StringRecordWriterProvider写入文本格式文档时，还可以指定gzip的压缩格式，扩展其它压缩格式很容易，只需要添加 两行代码就可以，建议增加lzo和lzop的压缩格式，以和我们Hive保持一致。

输出格文件类型：建议文本格式的文件

文件目录规则：（配置的目录）+ topic名 + daily|hour + （年/月/日）|（年/月/日/小时） + 数据文件，例如：/rocketmq/data/vip_ods_heartbeat/daily/2015/06/10 /vip_ods_heartbeat.broker-a.0.999.48388735.1433865600000.deflate

文件名规则：topic名+ （RocketBrokerId）|（kafka的对应分区的learder的BrokerId）+ （RocketQueueId）|（kafka分区号）+ 写入消息行数 + 最后一条消息的Offset + 编码的分区（时间 + 压缩格式后缀）,例如：vip_ods_heartbeat.broker- a.0.999.48388735.1433865600000.deflate

Topic的命名规则：业务标识+数据库名+数据库表名（分表只需要BaseName就可以），例 如：vip_ods_heartbeat

消息格式：操作类型\t表名（分表的话是分表名）\t数据库名\t主键名\t唯一索引\tBinlog日志时间 \tCheckPoint字段\tDataBefore\tDataAfter，库名表名都是RockMQ中的原始数据，在生成列数据时，列中数 据如果有\t等特殊字符需要替换，例如：
insert\theartbeat\tvip_ods\tid\tname,pid\t1232132131\t120@21\t{“字段名”,”字段 值”,…}\t{“字段名”,”字段值”,…}

确定数据导入是否完成：Camus中会在History的目录中存放历次消费的状态，包括开始执行的分区和它们的Offset、 执行结束位置的分区和它们的Offset，这两个文件以SequenceFile的形式存放在HDFS文件中，另外Camus在执行结束后可以把 执行信息汇总发送到Kafka的Topic中，Topic的名字为：TrackingMonitoringEvent，如果监控程序监控这个 Topic，是可以得到当前执行的情况的信息的。请评估一下这两种方式，看是否够用，有没有必要自行实现Checker机制。