sqoop内部结构

1.1  Sqoop内部结构

本节介绍Sqoop的内部结构。

Sqoop程序是由的主类com.cloudera.sqoop.Sqoop驱动。有限数量的额外的类在同一个包:SqoopOptions (如前所述),ConnFactory(即熟练操作ManagerFactory实例)。

1.1.1    一般程序流程

一般的程序流程如下:

com.cloudera.sqoop.Sqoop是主类和实现了Tool,一个新的实例ToolRunner被推出,Sqoop的第一个参数是一个字符串,并在SqoopTool中定义并执行,SqoopTool来执行用户的各种请求操作(如: import, export, codegen等)。

SqoopTool将解析其余的参数,设置相应的字段在SqoopOptions类,然后执行。

在SqoopTool 的run()方法中,import 、export或者其他正确的指令被执行。一般情况下ConnManager一般在SqoopOptions数据的基础上进行实例化。ConnFactory被用于从ManagerFactory中获取一个ConnManager。这在前面的部分已经进行了描述。Imports 、 exports或者其他大型数据的迁移操作通常是一个并行的、可靠的MapReduce任务。Import操作并不是一定要以MapReduce作业的方式运行,ConnManager.importTable()将确定如何以最佳的方式进行import操作。每一个主要操作实际上都有ConnMananger控制,除了生成代码的操作,这些操作是由CompilationManager和ClassWriter做的(都在 com.cloudera.sqoop.orm 包中)。导入数据到Hive的操作是由com.cloudera.sqoop.hive.HiveImport的importTable()完成的,这样做是为了不为使用ConnManager的实现都担心。

ConnManager 的 importTable()方法接收一个类型为ImportJobContext的参数,其中包含这个方法所需的各种参数值。将来,该类可扩展附加参数,以实现功能更加强大的import操作。同样, exportTable()方法接收的参数类型ExportJobContext。这些类包含一系列导入/导出,指向SqoopOptions对象,以及其他相关的数据。

1.1.2    子包

com.cloudera.sqoop包中的子包,包括:

※  Hive:便于数据导入到Hive

※  IO: 实现的java.io. *接口

※  Lib: 外部公共API(如前所述)

※  Manager: ConnManager和ManagerFactory的接口和它们的实现

※  Mapreduce: 新的(0.20 +)MapReduce的API接口的类

※  Orm: 代码自动生成

※  Tool: 实现SqoopTool

※  Util: 各种实用工具类

IO包中的OutputStream和BufferedWriter的实现被用于直接向HDFS进行写入。SplittableBufferedWriter允许为客户端打开一个单一的BufferedWriter,在引擎之下,当达到目标值后连续的写入文件系统。这允许在Sqoopimport的同时使用压缩机制(如gzip),同时允许在MapR任务之后对数据集进行分割。大对象文件存储系统的代码也存在于IO包中。

Mapreduce包中的代码用于直接与Hadoop MapReduce做接口,将在下一章节中详述。

ORM包中的代码用于生产代码。它依赖于提供了com.sun.tools.javac包的JDK的tools.jar包。

UTIL包中包含用于整个Sqoop的各种工具

※  ClassLoaderStack:管理由当前线程的堆栈使用的ClassLoader的实例,这正是当以本地模式运行MapReduce任务时自动生成代码写入当心线程的原理。

※  DirectImportUtils:包含一些直接进行Hadoop import操作的方便方法。

※  Executor:启动外部进程,并连接这些来生成由一个AsyncSink(见下面更详细地)的流处理程序。

※  ExportException:当exports失败时由ConnManagers抛出异常。

※  ImportException:当import失败时由ConnManagers抛出异常。

※  JDBCURL:处理连接字符串的解析,这是类URL,是不规范的、不合格的。

※  PerfCounters:被用来估计用于显示给用户的传输速率。

※  ResultSetPrinter:漂亮地打印结果集。

在不同的时候,Sqoop从外部程序中读取stdout,最简单的模式就是由LocalMySQLManager和DirectPostgresqlManager执行的直接模式 (direct-mode)import。之后由Runtime.exec()产生一个进程,该进程的标准输出(Process.getInputStream())和潜在错误(Process.getErrorStream())都需要被处理。在这些流之间无法读取更多的数据从而导致在写入更多数据之前外部进程阻塞。因此,这些都必须处理,最好以异步的方式。

按照Sqoop的说法,“异步接收器”是一个线程需要一个InputStream和读取完成。这些实现AsyncSink实现。com.cloudera.sqoop.util.AsyncSink抽象类定义了这家工厂必须执行的操作。processStream()将产生另一个线程立即开始处理从InputStream中读取的数据参数; 它必须读这流来完成的。 join()方法允许外部线程等待,直到处理完成。

一些"stock"被同步实现:LoggingAsyncSink将重复InputStream上的一切在log4j INFO语句中。NullAsyncSink消耗所有的输入和什么都不做。

各种ConnManagers使得外部进程以内部类的方式拥有它们自己的AsyncSink实现,它们通过数据库tools读取,并且促使数据流向HDFS,有可能在此期间进行格式转换。

1.1.3    与MapReduce的接口

Sqoop调度MapReduce作业产生imports和exports。配置和执行的MapReduce工作如下几个共同的步骤(配置InputFormat配置OutputFormat设置映射的实施;等等)。这些步骤是在com.cloudera.sqoop.mapreduce.JobBase类中的。为了使用,JobBase允许一个用户来指定InputFormat,OutputFormat,和映射器。

JobBase本身就是ImportJobBase和ExportJobBase的子集,为特定的配置步骤提供更好的支持,分别由ImportJobBase和ExportJobBase的子类。 ,ImportJobBase.runImport()将调用的配置步骤,并运行一个工作表导入HDFS。

更多精彩内容请关注:http://bbs.superwu.cn

关注超人学院微信二维码:

时间: 2024-08-14 01:56:08

sqoop内部结构的相关文章

解决sqoop报错Invalid number; item = ITEM_UNICODE

报错栈: java.sql.SQLException: Invalid number; item = ITEM_UNICODE at com.intersys.jdbc.SysList.getInt(SysList.java:1735) at com.intersys.jdbc.CacheResultSet.getInt(CacheResultSet.java:247) at org.apache.sqoop.lib.JdbcWritableBridge.readInteger(JdbcWrit

sqoop同步mysql到hdfs

链接:http://pan.baidu.com/s/1gfHnaVL 密码:7j12 mysql-connector version 5.1.32 若在安装版本过程遇到些问题,可参考http://dbspace.blog.51cto.com/6873717/1875955,其中一些问题的解决办法 下载并安装: cd /usr/local/tar -zxvf sqoop2-1.99.3-cdh5.0.0.tar.gzmv sqoop2-1.99.3-cdh5.0.0 sqoop添加sqoop2到系

解决sqoop报错:SQLServerException: 将字符串转换为 uniqueidentifier 时失败。

报错栈: Error: java.io.IOException: Cannection handler cannot recover failure: at org.apache.sqoop.mapreduce.db.SQLServerDBRecordReader.nextKeyValue(SQLServerDBRecordReader.java:169) at org.apache.hadoop.mapred.MapTask$NewTrackingRecordReader.nextKeyVal

sqoop 使用笔记

好久没有更新自己技术博客,现在开始工作了,把自己遇到的问题写到这里边来 主要把自己的问题写出来,分享给大家 sqoop 导入数据时候 有时候会遇到mysql 中有sql 中的关键字 这时候如果直接导出到hive 中会出现错误 例如下面的例子 这时候会出现错误 可以使用--query 来解决问题 例如 : sqoop-import -D mapreduce.map.memory.mb=256 -D mapreduce.map.java.opts=-Xmx128m  --connect ''--us

Sqoop 脚本开发规范(实例手把手带你写sqoop export和sqoop import)

首先,先明确,为什么Sqoop需要规范的脚本开发呢? 答:是因为,Sqoop import HDFS/Hive/HBase这些都是手动.但是在实际生产里,有时候,需要用脚本来完成. 比如,通过shell脚本来操作对Sqoop.Hive.HBase.MapReduce.HDFS.Spark.Storm等各种. Sqoop 脚本开发规范 目录规范 1.目录结构体系 /home/hadoop(开发用户)/app/djt(数据来源.业务)/sh/sqoop 示例:/home/hadoop/app/djt

Apache的HBase与cdh的sqoop集成(不建议不同版本之间的集成)

1.修改sqoop的配资文件 2.从mysql导入到hbase(import) bin/sqoop import \ --connect jdbc:mysql://linux-hadoop3.ibeifeng.com:3306/sqoop \ --username root \ --password 123456 \ --table tohdfs \ --hbase-create-table \ --hbase-table s1 \ --hbase-row-key id \ --column-f

[Sqoop]Sqoop使用

Sqoop的本质还是一个命令行工具,和HDFS,MapReduce相比,并没有什么高深的理论. 我们可以通过sqoop help命令来查看sqoop的命令选项,如下: 16/11/13 20:10:17 INFO sqoop.Sqoop: Running Sqoop version: 1.4.6 usage: sqoop COMMAND [ARGS] Available commands:  codegen            Generate code to interact with da

Innodb Buffer Pool内部结构

Innodb Buffer Pool内部结构 1.    Innodb Buffer 功能 Innodb buffer pool的主要功能存储外存页面在内存中的镜像.镜像有如下2种镜像: (1)只读镜像:只读镜像读取的是非脏页. (2)更新镜像:更新镜像为buffer pool中的脏页. Innodb实现了行级多版本(MVCC),而不是整个页的多版本.Oracle在实现中存在第三种镜像,就是版本镜像.而innodb中确实没有.在innodb中任何外存中的脏页的读取以及更新都是在buffer po

sqoop job local 和 Cannot initialize Cluster 问题

hadoop版本:Hadoop 2.3.0-cdh5.0.0 sqoop版本:Sqoop 1.4.4-cdh5.0.0 配置好sqooop-env.xml: #Set path to where bin/hadoop is available export HADOOP_COMMON_HOME=/my/hadoop #Set path to where hadoop-*-core.jar is available export HADOOP_MAPRED_HOME=/my/hadoop/shar