Sqoop开发者指南

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

2、V1.4.2 o3 Sqoop发行版Apache的Sqoop是Apache软件基金会的一个开放源码的软件产品。Sqoop产品的发展在： http:/s vn ./repos/asf/sqoop/tru nk。在该网站上，你可以得到：探最新版本的源代码An issue tracker探 包含 WIKI的Sqoop文档4先决条件Sqoop开发需要如下先决知识：探JAVA开发：&熟练掌握JDBC&熟练掌握Hadoop的API （包括新的Map Reduce 0.2版本以上的API）探关系型数据库和SQL本文档假定您使用的是Linux或类似Linux的环境。如果你使用的是 Wind

3、ows 您可能能够使用cygwin完成以下任务。如果你使用的是 Mac OS X你应该看到一 些兼容性错误（如果有的话）。 Sqoop主要在Linux环境上进行测试。5 Sqoop源码您可以从如下网站上获取Sqoop的源代码：http:/s vn. /repos/asf/sqoop/tr unkSqoop源代码被保存在一个“ Git ”仓库中，从仓库中获取 Sqoop源码的操作 说明被提供在如上网站中提供的一个“ T0DO页面上。编译指令由源码根目录下的“ COMPILING.txt ”文件提供。6开发者API参考本章节的内容提供给那些需要整合或者扩展Sqoop的应用程序开

4、发者。下面三个部分都是为了接下来的用例而写的：探使用Sqoop生成的类和公共类探Sqoop扩展（即，于更多数据库交互的额外的ConnManage的实现）探修改Sqoop内部实现每一部分逐步深入的描述Sqoop系统。6.1外部APISqoop在执行的时候自动生成将关系型数据库里表导入（import ）到HDFS系 统中的java类，这个类包含导入表的每一列的成员字段，这个类的一个实例保存 导入表的每一行（每一行为一个实例），这个类通过Hadoop实现序列化API，命名为 Writable 和DBWritable 接口。并且包含其他比较方便的方法：探parse（）:可以解析带分隔符文本的方法。探t

5、oString（）:可以保留用户选择的分隔符的方法 确保自动生成的类中的所有方法都存在于下面的抽象类中： com.cloudera.sqoop.lib.SqoopRecord.SqoopRecord实例可能依赖于Sqoop的公共API。所有的类都存在于 com.cloudera.sqoop.lib 包中。这些被简要的描述如下。Sqoop客户端并不需要直 接与这些类发生关系，尽管由Sqoop生成的类依赖于他们。因此，这些AOP被认为 是公开的，并且需要继续优化的。探RecordParser类将解析文本文件中的一行到一个字段列表中，使用可控的 分隔符和引号字符。探静态类 FieldFormatte

6、r 提供了一个被用于 SqoopRecord.toString()的方 法，该方法处理引用和转义字符。探 圭寸装 ResultSet、PreparedStatement objects、SqoopRecords之间的数据 是通过 JdbcWritableBridge 实现的。探 BigDecimalSerializer 在 Writable 接 口之上包含了一组对 BigDecimal (小 数)对象序列化的方法。6.2扩展API6.2.1 HBase的序列化扩展本章节包含了 API和最基本的用于扩展Sqoop的类，这些类使得Sqoop可以与 更多的数据库提供商做接口。Sqoop使用JDBC和

7、DataDrivenDBInputFormat的从数据库中读取数据，对不 同的数据库提供商及JDBC元数据之间的不同，必须为大多数数据库提供特定的代 码路径。Sqoop针对这个问题的解决方案是引入 ConnManager API (com.cloudera.sqoop.ma nager.C onnManan ger)。Co nnMan agei是 一个静态类，这个类定义了数据库本身相关作用的方法。ConnMan ager类的大多数实现都继承静态类 SqlMa nager(com.cloudera.sqoop.manager.SqlManager ) ，SqlManager 使用标准的 SQL执

8、行 大部分的动作。所有的子类需要实现getConnection()方案，用于获取实际的JDBC 数据库链接，子类可以重写其他所有方法。SqlMa nager类本身暴露了一个受保护 的API，使开发人员可以有选择地重写行为。例如getColNamesQuery()方法允许使 用getColNames()的进行修改SQL查询，而不需要重写大多数的 getColNames()方 法。ConnManage的实现通过Sqoop特定的类(SqoopOptions)获得了大量的配置 数据，SqoopOptions是可变的。SqoopOptions不直接存储特定的管理信息， 相反, 他包含了对配置信息的一个指

9、向，通过Ge nericOptio nsParser解析命令行参数之后由 Tool.getC onf() 返回。这使得通过“ -D any .specific.param= any .value ” 扩展参数，而不需要任何层次的的解析参数或修改SqoopOptio ns。配置信息中的基础配置通过调用工作流交给任意的 MapReduce作业，所以，用户可以在命令行自 定义设置必要的Hadoop状态。目前所有的Conn Ma nage实现都是无状态的，因此，Co nn Ma nagers示例的系 统在Sqoop的生命周期中对同一个ConnManage类实现了多个实例。目前实例化一 个ConnMan

10、agers类是一个轻量级的操作，并且很少这样做。因此 ConnManagers 不对operations之间进行缓存。当前的Conn Ma nagers是由静态类 Man agerFactory的一个实例创建的(参考： http:/issues. /jira/browse/MAPREDUCE-750 )。当前的 ManagerFactory的一个实现服务于整个Sqoop： com.cloudera.sqoop.manager.DefaultManagerFactory。Sqoop 的相关扩展不应该修改DefaultManagerFactory 。相反，ManagerFact

11、ory的一个特定的扩展需要由新 的ConnManage提供。值得注意的是ManagerFactory有一个单一的名为accept() 的方法，这个方法将确定是否可以实例化为用于用户输入信息(user sSqoopOptions )的ConnManage。如果可以，将返回一个 ConnManagei实例，否则 返回null。ManagerFactory实例的使拥由设置在 sqoop-site.xml 中的sqoop.co nn ectio n.factories支配，扩展库的用户可以安装第三方库，其中包含一个新的 ManagerFactory 和 ConnManager(S)，配置 sqoop-

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

13、存放到单元格内所有这些都通过 HBase客户端API中的Put statements实现。Sqoop于HBase 的交互执行都在com.cloudera.sqoop.hbase 包里。记录从数据库中被 deserialzed，并从mapper任务中发出。OutputFormat负责把结果插入到HBase 中，这是通过一个叫做 PutTransformer的接口实现的。PutTransformer有一个方 法称为getPutComman（），它使用一个MapString,对象等领域的数据集作为输 入。它返回列表,描述了如何将单元格插入到 HBase中。默认的 PutTransformer实现是T

14、oStringPutTransformer ，使用基于字符串的方式将每个 字段序列化到HBase中。你可以通过自己的PutTransformer重新这个实现，并且把他添加到 map任务 的classpath中，来告诉Sqoop你将使用你自己的实现，通过” -D”命令设置自己 的 class:sqoop.hbase.insert.put.transformer.class。在您的 PutTransformer 实现，特定的 row key、column、column family 可通过getRowKeyColumn（）和getColumnFamily （）方法。你可以扩展更多的超 出限制的”

15、PUT操作，例如，注入额外的辅助索引的行。Sqoop将对-hbase-table 命令特定的表执行所有的” PUT操作。6.3 Sqoop内部结构本节介绍Sqoop的内部结构。Sqoop程序是由的主类com.cloudera.sqoop.Sqoop驱动。有限数量的额外的 类在同一个包：SqoopOptions（如前所述），ConnFactory （即熟练操作ManagerFactory 实例）。6.3.1一般程序流程般的程序流程如下:com.cloudera.sqoop.Sqoop 是主类和实现了 Tool, 一个新的实例 TooIRunner 被推出，Sqoop的第一个参数是一个字符串，并在

16、 SqoopTool中定义并执行， SqoopTool来执行用户的各种请求操作(如： import, export, codegen 等)。SqoopTool将解析其余的参数，设置相应的字段在SqoopOptions类，然后执行。在SqoopTool的run()方法中，import、export或者其他正确的指令被执 行。一般情况下ConnManager一般在SqoopOptions数据的基础上进行实例化。 Conn Factory被用于从 Ma nagerFactory中获取一个 Co nnMan age。这在前面的部分 已经进行了描述。Imports、exports或者其他大型数据的迁移操

17、作通常是一个 并行的、可靠的 Map Reduced务。Import操作并不是一定要以 Map Reducd乍业的 方式运行，Co nnMan ager.importTable ()将确定如何以最佳的方式进行 import 操作。每一个主要操作实际上都有 Co nnManan ger控制，除了生成代码的操作，这 些操作是由CompilationManager和ClassWriter 做的(都在 com.cloudera.sqoop.orm 包中)。导入数据到Hive的操作是由 com.cloudera.sqoop.hive.Hivelmport 的 importTable()完成的，这样做是为

18、了不 为使用ConnManager的实现都担心。ConnMan ager 的 importTable ()方法接收一个类型为 ImportJobC on text 的参数，其中包含这个方法所需的各种参数值。将来，该类可扩展附加参数，以实 现功能更加强大的import操作。同样， exportTable ()方法接收的参数类型 ExportJobContext 。这些类包含一系列导入/导出，指向SqoopOptions对象，以及 其他相关的数据。6.3.2 子包com.cloudera.sqoop 包中的子包，包括：探Hive :便于数据导入到 Hive探IO:实现的java.io. * 接口探

19、Lib:外部公共API (如前所述)探 Manager: ConnManager和 ManagerFactory 的接口和它们的实现探 Map reduce:新的(0.20 +)Map Reduces API 接口的类探Orm:代码自动生成探 Tool:实现 SqoopTool探Util:各种实用工具类IO包中的Outputstream和BufferedWriter 的实现被用于直接向 HDFS进行写 入。SplittableBufferedWriter允许为客户端打开一个单一的 BufferedWriter ，在引擎之下，当达到目标值后连续的写入文件系统。这允许在Sqoop import的同

20、时使用压缩机制(如gzip),同时允许在MapR任务之后对数据集进行分割。大对 象文件存储系统的代码也存在于IO包中。Map reduce包中的代码用于直接与 Hadoop Map Reduces接口，将在下一章节 中详述。OR包中的代码用于生产代码。它依赖于提供了 com.sun.tools.javac 包的JDK 的 tools.jar 包。UTIL包中包含用于整个Sqoop的各种工具探ClassLoaderStack :管理由当前线程的堆栈使用的 ClassLoader的实例， 这正是当以本地模式运行 MapReduce壬务时自动生成代码写入当心线程的 原理。探DirectImportU

21、tils:包含一些直接进行 Hadoop import操作的方便方法。探Executor :启动外部进程，并连接这些来生成由一个 AsyncSink (见下面 更详细地)的流处理程序。探 ExportException :当 exports 失败时由 ConnManagers抛出异常。探 ImportException :当 import 失败时由 ConnManagers抛出异常。探JDBCURL处理连接字符串的解析，这是类 URL是不规范的、不合格的。探PerfCounters :被用来估计用于显示给用户的传输速率。探ResultSetPrinter :漂亮地打印结果集。在不同的时候，Sq

22、oop从外部程序中读取stdout，最简单的模式就是由 LocalMySQLManage和 DirectPostgresqlManager 执行的直接模式 (direct-mode ) import。之后由Runtime.exec()产生一个进程，该进程的标准输出(Process.getInputStream()和潜在错误(Process.getErrorStream() 都需要被 处理。在这些流之间无法读取更多的数据从而导致在写入更多数据之前外部进程阻 塞。因此，这些都必须处理，最好以异步的方式。按照Sqoop的说法，“异步接收器”是一个线程需要一个InputStream和读取 完成。这些实

23、现 AsyncSink 实现。com.cloudera.sqoop.util.AsyncSink抽象类定义了这家工厂必须执行的操作。processStream ()将产生另一个线程立即开始处 理从InputStream中读取的数据参数；它必须读这流来完成的。join ()方法允 许外部线程等待，直到处理完成。一些stock被同步实现：LoggingAsyncSink将重复InputStream 上的一切在 Iog4j INFO 语句中。NullAsyncSink消耗所有的输入和什么都不做。各种ConnManagers使得外部进程以内部类的方式拥有它们自己的 AsyncSink 实现，它们通过数

24、据库tools读取，并且促使数据流向HDFS有可能在此期间进 行格式转换。633 与 MapReduce 的接口Sqoop调度 MapReducd业产生 imports 和 exports。配置和执行的 MapReduce 工作如下几个共同的步骤(配置InputFormat配置OutputFormat设置映射的实施； 等等)。这些步骤是在 com.cloudera.sqoop.mapreduce.JobBase 类中的。 为了使 用，JobBase允许一个用户来指定InputFormat ， OutputFormat，和映射器。JobBase本身就是ImportJobBase和ExportJo

25、bBase的子集，为特定的配置步 骤提供更好的支持，分别由ImportJobBase和ExportJobBase的子类。， ImportJobBase.ru nlmport()将调用的配置步骤，并运行一个工作表导入HDFS7 Eclispe 集成1、从官方网站上下载：sqoop-142.bin_hadoop-0.23.tar.gz2、解压，用ant编译，得到相关jar包3、 新建一个web工程，将上面得到的jar包copy到lib目录下，并添加数据 库驱动包(ojdbc6.jar )4、将sqoop的java源码添加到src目录下8 Eclispe 远程调试 sqoop利用eclipse远程调试功能，