Sqoop开发者指南

1、Sqoop开发者指南目录1 . 引言12支持的发行版 .13Sqoop 发行版 .14先决条件 .1I5Sqoop 源码.26开发者 API 参考 .26.1外部 API .26.2扩展 API .36.2.1的.错误！未定义书签。6.3Sqoop 内部实现 .56.3.1 .56.3.2的.错误！未定义书签。6.3.3Hadoop 大数据处理技术架构 .错误！未定义书签。II1 引言如果你是一个开发者或应用程序的程序员打算修改的 Sqoop或通过使用 Sqoop 内部 API 进行扩展，你应该阅读此文档。 以下各节描述的每一个 API 的作用，包括内部 API 和支持其他数据库必须的 API

2、。2 支持的发行版本文档适用于 Sqoop V1.4.2 。3 Sqoop 发行版Apache的 Sqoop是 Apache软件基金会的一个开放源码的软件产品。Sqoop 产品的发展在： /repos/asf/sqoop/trunk。在该网站上，你可以得到： 最新版本的源代码 An issue tracker 包含 WIKI 的 Sqoop 文档4 先决条件Sqoop开发需要如下先决知识： JAVA开发：熟练掌握 JDBC熟练掌握 Hadoop的 API（包括新的 MapReduce 0.2 版本以上的 API）1 关系型数据库和 SQL本文档假定您使用

3、的是Linux 或类似 Linux 的环境。如果你使用的是Windows，您可能能够使用cygwin 完成以下任务。如果你使用的是Mac OS X，你应该看到一些兼容性错误（如果有的话）。Sqoop 主要在 Linux 环境上进行测试。5 Sqoop 源码您可以从如下网站上获取Sqoop的源代码：/repos/asf/sqoop/trunkSqoop源代码被保存在一个“ Git ”仓库中，从仓库中获取 Sqoop源码的操作说明被提供在如上网站中提供的一个“ TODO”页面上。编译指令由源码根目录下的“COMPILING.txt ”文件提供。6 开发者 A

4、PI 参考本章节的内容提供给那些需要整合或者扩展Sqoop 的应用程序开发者。下面三个部分都是为了接下来的用例而写的： 使用 Sqoop生成的类和公共类 Sqoop扩展（即，于更多数据库交互的额外的ConnManager的实现） 修改 Sqoop内部实现每一部分逐步深入的描述Sqoop系统。6.1 外部 APISqoop在执行的时候自动生成将关系型数据库里表导入（import ）到 HDFS系统中的 java 类，这个类包含导入表的每一列的成员字段，这个类的一个实例保存导入表的每一行（每一行为一个实例），这个类通过Hadoop实现序列化 API，命名为 Writable和 DBWritable

5、接口。并且包含其他比较方便的方法： parse() ：可以解析带分隔符文本的方法。 toString() ：可以保留用户选择的分隔符的方法2确保自动生成的类中的所有方法都存在于下面的抽象类中：com.cloudera.sqoop.lib.SqoopRecord.SqoopRecord实例可能依赖于 Sqoop的公共 API。所有的类都存在于 com.cloudera.sqoop.lib 包中。这些被简要的描述如下。 Sqoop 客户端并不需要直接与这些类发生关系， 尽管由 Sqoop生成的类依赖于他们。 因此，这些 AOP被认为是公开的，并且需要继续优化的。 RecordParser 类将解析

6、文本文件中的一行到一个字段列表中， 使用可控的分隔符和引号字符。 静态类 FieldFormatter 提供了一个被用于 SqoopRecord.toString() 的方法，该方法处理引用和转义字符。 封装 ResultSet 、PreparedStatement objects 、SqoopRecords 之间的数据是通过 JdbcWritableBridge 实现的。 BigDecimalSerializer 在 Writable 接口之上包含了一组对 BigDecimal（小数）对象序列化的方法。6.2 扩展 API6.2.1HBase 的序列化扩展本章节包含了 API 和最基本的用于

7、扩展Sqoop的类，这些类使得 Sqoop 可以与更多的数据库提供商做接口。Sqoop使用 JDBC和 DataDrivenDBInputFormat 的从数据库中读取数据，对不同的数据库提供商及 JDBC元数据之间的不同，必须为大多数数据库提供特定的代码路径。 Sqoop针对这个问题的解决方案是引入 ConnManager API(com.cloudera.sqoop.manager.ConnMananger)。ConnManager是一个静态类，这个类定义了数据库本身相关作用的方法。ConnManager类的大多数实现都继承静态类SqlManager（ com.cloudera.sqoop

8、.manager.SqlManager ），SqlManager 使用标准的 SQL执行大部分的动作。 所有的子类需要实现 getConnection() 方案，用于获取实际的 JDBC数据库链接，子类可以重写其他所有方法。SqlManager 类本身暴露了一个受保护的 API，使开发人员可以有选择地重写行为。例如 getColNamesQuery() 方法允许使3用 getColNames() 的进行修改 SQL查询，而不需要重写大多数的getColNames() 方法。ConnManager的实现通过 Sqoop特定的类（ SqoopOptions ）获得了大量的配置数据，SqoopOpt

9、ions 是可变的。 SqoopOptions 不直接存储特定的管理信息，相反，他包含了对配置信息的一个指向，通过GenericOptionsParser解析命令行参数之后由 Tool.getConf() 返回。这使得通过“ - D any.specific.param= any.value ” 扩展参数，而不需要任何层次的的解析参数或修改 SqoopOptions 。配置信息中的基础配置通过调用工作流交给任意的 MapReduce作业，所以，用户可以在命令行自定义设置必要的 Hadoop状态。目前所有的 ConnManager实现都是无状态的，因此，ConnManagers示例的系统在 Sq

10、oop 的生命周期中对同一个ConnManager类实现了多个实例。目前实例化一个 ConnManagers类是一个轻量级的操作，并且很少这样做。因此ConnManagers不对 operations之间进行缓存。当前的 ConnManagers是由静态类 ManagerFactory 的一个实例创建的（参考：/jira/browse/MAPREDUCE-750）。当前的ManagerFactory 的一个实现服务于整个Sqoop：com.cloudera.sqoop.manager.DefaultManagerFactory。Sqoop 的相关扩

11、展不应该修改 DefaultManagerFactory。相反，ManagerFactory 的一个特定的扩展需要由新的 ConnManager提供。值得注意的是 ManagerFactory 有一个单一的名为 accept()的方法，这个方法将确定是否可以实例化为用于用户输入信息（user sSqoopOptions ）的 ConnManager。如果可以，将返回一个ConnManager实例，否则返回 null 。ManagerFactory 实例的使拥由设置在sqoop-site.xml中的sqoop.connection.factories支配，扩展库的用户可以安装第三方库，其中包含一

12、个新的 ManagerFactory 和 ConnManager（ S），配置 sqoop-site.xml使用新的ManagerFactory 。DefaultManagerFactory principly通过存储在 SqoopOptions中的连接字符串区分数据库。扩展者可以利用 MapReduce，com.cloudera.sqoop.io和 util包中的类以方便他们的扩展实现。下面一节中更详细地描述这些包和类。4Sqoop支持从关系型数据库到HBase的导入操作。当导入数据到 HBase的时候，必须转化成 HBase可以接受的特定的形式： 数据必须放到 HBase中的一个或多个表中

13、。 输入数据的列必须到放置到一个column family钟。 值必须序列化成字节数组存放到单元格内所有这些都通过 HBase客户端 API 中的 Put statements 实现。Sqoop于 HBase的交互执行都在com.cloudera.sqoop.hbase包里。记录从数据库中被deserialzed ，并从 mapper 任务中发出。 OutputFormat 负责把结果插入到 HBase 中，这是通过一个叫做 PutTransformer 的接口实现的。 PutTransformer 有一个方法称为 getPutCommand（），它使用一个 Map等领域的数据集作为输入。它返

14、回列表 ，描述了如何将单元格插入到 HBase中。默认的 PutTransformer 实现是 ToStringPutTransformer ，使用基于字符串的方式将每个字段序列化到 HBase中。你可以通过自己的PutTransformer重新这个实现，并且把他添加到map任务的 classpath中，来告诉 Sqoop你将使用你自己的实现， 通过” -D”命令设置自己的 class:sqoop.hbase.insert.put.transformer.class。在您的 PutTransformer实现，特定的 row key、column 、 column family可通过 getRo

15、wKeyColumn（）和 getColumnFamily （）方法。你可以扩展更多的超出限制的” PUT”操作，例如，注入额外的辅助索引的行。 Sqoop 将对 -hbase-table 命令特定的表执行所有的” PUT”操作。6.3 Sqoop 内部结构本节介绍 Sqoop 的内部结构。Sqoop程序是由的主类 com.cloudera.sqoop.Sqoop 驱动。有限数量的额外的类在同一个包： SqoopOptions （如前所述）， ConnFactory （即熟练操作 ManagerFactory 实例）。6.3.1一般程序流程一般的程序流程如下：5com.cloudera.sqo

16、op.Sqoop是主类和实现了Tool ，一个新的实例 ToolRunner被推出， Sqoop的第一个参数是一个字符串，并在SqoopTool 中定义并执行，SqoopTool 来执行用户的各种请求操作（如：import, export, codegen等）。SqoopTool 将解析其余的参数，设置相应的字段在SqoopOptions 类，然后执行。在 SqoopTool 的 run() 方法中， import 、 export 或者其他正确的指令被执行。一般情况下 ConnManager一般在 SqoopOptions 数据的基础上进行实例化。ConnFactory 被用于从 Manag

17、erFactory 中获取一个 ConnManager。这在前面的部分已经进行了描述。 Imports 、 exports 或者其他大型数据的迁移操作通常是一个并行的、可靠的 MapReduce任务。 Import 操作并不是一定要以 MapReduce作业的方式运行， ConnManager.importTable （）将确定如何以最佳的方式进行 import操作。每一个主要操作实际上都有ConnMananger控制，除了生成代码的操作，这些操作是由 CompilationManager 和 ClassWriter 做的 ( 都在 com.cloudera.sqoop.orm 包中 ) 。导

18、入数据到 Hive 的操作是由com.cloudera.sqoop.hive.HiveImport的 importTable()完成的，这样做是为了不为使用 ConnManager的实现都担心。ConnManager 的importTable （）方法接收一个类型为ImportJobContext的参数，其中包含这个方法所需的各种参数值。将来，该类可扩展附加参数，以实现功能更加强大的import 操作。同样，exportTable （）方法接收的参数类型ExportJobContext 。这些类包含一系列导入 / 导出，指向 SqoopOptions 对象，以及其他相关的数据。6.3.2子包c

19、om.cloudera.sqoop包中的子包，包括： Hive ：便于数据导入到 Hive IO:实现的 java.io. *接口 Lib: 外部公共 API（如前所述） Manager: ConnManager 和 ManagerFactory 的接口和它们的实现 Mapreduce: 新的（ 0.20 + ）MapReduce的 API 接口的类6 Orm: 代码自动生成 Tool: 实现 SqoopTool Util: 各种实用工具类IO 包中的 OutputStream 和 BufferedWriter的实现被用于直接向HDFS进行写入。 SplittableBufferedWrite

20、r允许为客户端打开一个单一的BufferedWriter，在引擎之下，当达到目标值后连续的写入文件系统。这允许在Sqoop import的同时使用压缩机制（如 gzip ）, 同时允许在 MapR任务之后对数据集进行分割。大对象文件存储系统的代码也存在于 IO 包中。Mapreduce包中的代码用于直接与Hadoop MapReduce做接口，将在下一章节中详述。ORM包中的代码用于生产代码。 它依赖于提供了 com.sun.tools.javac包的 JDK的 tools.jar包。UTIL 包中包含用于整个Sqoop的各种工具 ClassLoaderStack ：管理由当前线程的堆栈使用的

21、 ClassLoader 的实例，这正是当以本地模式运行 MapReduce任务时自动生成代码写入当心线程的原理。 DirectImportUtils ：包含一些直接进行 Hadoop import 操作的方便方法。 Executor ：启动外部进程，并连接这些来生成由一个AsyncSink （见下面更详细地）的流处理程序。 ExportException ：当 exports 失败时由 ConnManagers抛出异常。 ImportException ：当 import 失败时由 ConnManagers抛出异常。 JDBCURL：处理连接字符串的解析，这是类 URL，是不规范的、不合格的

22、。 PerfCounters ：被用来估计用于显示给用户的传输速率。 ResultSetPrinter ：漂亮地打印结果集。在不同的时候， Sqoop从外部程序中读取stdout ，最简单的模式就是由LocalMySQLManager和 DirectPostgresqlManager执行的直接模式 （direct-mode ）import 。之后由 Runtime.exec() 产生一个进程，该进程的标准输出(Process.getInputStream()和潜在错误 (Process.getErrorStream()都需要被7处理。在这些流之间无法读取更多的数据从而导致在写入更多数据之前外部

23、进程阻塞。因此，这些都必须处理，最好以异步的方式。按照 Sqoop 的说法，“异步接收器”是一个线程需要一个InputStream和读取完成。这些实现 AsyncSink 实现。 com.cloudera.sqoop.util.AsyncSink抽象类定义了这家工厂必须执行的操作。processStream （）将产生另一个线程立即开始处理从 InputStream中读取的数据参数 ;它必须读这流来完成的。join（）方法允许外部线程等待，直到处理完成。一些 stock 被同步实现： LoggingAsyncSink 将重复 InputStream 上的一切在log4j INFO语句中。 Nu

24、llAsyncSink消耗所有的输入和什么都不做。各种 ConnManagers使得外部进程以内部类的方式拥有它们自己的 AsyncSink 实现，它们通过数据库 tools 读取，并且促使数据流向 HDFS，有可能在此期间进行格式转换。6.3.3与 MapReduce 的接口Sqoop调度 MapReduce作业产生 imports 和 exports 。配置和执行的 MapReduce 工作如下几个共同的步骤 （配置 InputFormat 配置 OutputFormat 设置映射的实施 ; 等等）。这些步骤是在 com.cloudera.sqoop.mapreduce.JobBase 类中的。 为了使用， JobBase 允许一个用户来指定 InputFormat ， OutputFormat ，和映射器。JobBase 本身就是 ImportJobBase 和 ExportJobBase 的子集，为特定的配置步骤提供更好的支持，分别由 ImportJobBase 和 ExportJobBase 的子类。 ， ImportJobBase.runImport （）将调用的配置步骤，并运行一个工作表导入 HD