Sqoop supports imports from databases to HBase. When copying data into HBase, it must be transformed into a format HBase can accept. Specifically:

All of this is done via Put statements in the HBase client API. Sqoop’s interaction with HBase is performed in the com.cloudera.sqoop.hbase package. Records are deserialzed from the database and emitted from the mapper. The OutputFormat is responsible for inserting the results into HBase. This is done through an interface called PutTransformer. The PutTransformer has a method called getPutCommand() that takes as input a Map<String, Object> representing the fields of the dataset. It returns a List<Put> describing how to insert the cells into HBase. The default PutTransformer implementation is the ToStringPutTransformer that uses the string-based representation of each field to serialize the fields to HBase.

You can override this implementation by implementing your own PutTransformer and adding it to the classpath for the map tasks (e.g., with the -libjars option). To tell Sqoop to use your implementation, set the sqoop.hbase.insert.put.transformer.class property to identify your class with -D.

Within your PutTransformer implementation, the specified row key column and column family are available via the getRowKeyColumn() and getColumnFamily() methods. You are free to make additional Put operations outside these constraints; for example, to inject additional rows representing a secondary index. However, Sqoop will execute all Put operations against the table specified with --hbase-table.

The general program flow is as follows:

com.cloudera.sqoop.Sqoop is the main class and implements Tool. A new instance is launched with ToolRunner. The first argument to Sqoop is a string identifying the name of a SqoopTool to run. The SqoopTool itself drives the execution of the user’s requested operation (e.g., import, export, codegen, etc).

The SqoopTool API is specified fully in SIP-1.

The chosen SqoopTool will parse the remainder of the arguments, setting the appropriate fields in the SqoopOptions class. It will then run its body.

Then in the SqoopTool’s run() method, the import or export or other action proper is executed. Typically, a ConnManager is then instantiated based on the data in the SqoopOptions. The ConnFactory is used to get a ConnManager from a ManagerFactory; the mechanics of this were described in an earlier section. Imports and exports and other large data motion tasks typically run a MapReduce job to operate on a table in a parallel, reliable fashion. An import does not specifically need to be run via a MapReduce job; the ConnManager.importTable() method is left to determine how best to run the import. Each main action is actually controlled by the ConnMananger, except for the generating of code, which is done by the CompilationManager and ClassWriter. (Both in the com.cloudera.sqoop.orm package.) Importing into Hive is also taken care of via the com.cloudera.sqoop.hive.HiveImport class after the importTable() has completed. This is done without concern for the ConnManager implementation used.

A ConnManager’s importTable() method receives a single argument of type ImportJobContext which contains parameters to the method. This class may be extended with additional parameters in the future, which optionally further direct the import operation. Similarly, the exportTable() method receives an argument of type ExportJobContext. These classes contain the name of the table to import/export, a reference to the SqoopOptions object, and other related data.

The following subpackages under com.cloudera.sqoop exist:

The io package contains OutputStream and BufferedWriter implementations used by direct writers to HDFS. The SplittableBufferedWriter allows a single BufferedWriter to be opened to a client which will, under the hood, write to multiple files in series as they reach a target threshold size. This allows unsplittable compression libraries (e.g., gzip) to be used in conjunction with Sqoop import while still allowing subsequent MapReduce jobs to use multiple input splits per dataset. The large object file storage (see SIP-3) system’s code lies in the io package as well.

The mapreduce package contains code that interfaces directly with Hadoop MapReduce. This package’s contents are described in more detail in the next section.

The orm package contains code used for class generation. It depends on the JDK’s tools.jar which provides the com.sun.tools.javac package.

The util package contains various utilities used throughout Sqoop:

In several places, Sqoop reads the stdout from external processes. The most straightforward cases are direct-mode imports as performed by the LocalMySQLManager and DirectPostgresqlManager. After a process is spawned by Runtime.exec(), its stdout (Process.getInputStream()) and potentially stderr (Process.getErrorStream()) must be handled. Failure to read enough data from both of these streams will cause the external process to block before writing more. Consequently, these must both be handled, and preferably asynchronously.

In Sqoop parlance, an "async sink" is a thread that takes an InputStream and reads it to completion. These are realized by AsyncSink implementations. The com.cloudera.sqoop.util.AsyncSink abstract class defines the operations this factory must perform. processStream() will spawn another thread to immediately begin handling the data read from the InputStream argument; it must read this stream to completion. The join() method allows external threads to wait until this processing is complete.

Some "stock" AsyncSink implementations are provided: the LoggingAsyncSink will repeat everything on the InputStream as log4j INFO statements. The NullAsyncSink consumes all its input and does nothing.

The various ConnManagers that make use of external processes have their own AsyncSink implementations as inner classes, which read from the database tools and forward the data along to HDFS, possibly performing formatting conversions in the meantime.

Sqoop schedules MapReduce jobs to effect imports and exports. Configuration and execution of MapReduce jobs follows a few common steps (configuring the InputFormat; configuring the OutputFormat; setting the Mapper implementation; etc…). These steps are formalized in the com.cloudera.sqoop.mapreduce.JobBase class. The JobBase allows a user to specify the InputFormat, OutputFormat, and Mapper to use.

JobBase itself is subclassed by ImportJobBase and ExportJobBase which offer better support for the particular configuration steps common to import or export-related jobs, respectively. ImportJobBase.runImport() will call the configuration steps and run a job to import a table to HDFS.

Subclasses of these base classes exist as well. For example, DataDrivenImportJob uses the DataDrivenDBInputFormat to run an import. This is the most common type of import used by the various ConnManager implementations available. MySQL uses a different class (MySQLDumpImportJob) to run a direct-mode import. Its custom Mapper and InputFormat implementations reside in this package as well.

Sqoop parses the arguments which are passed by users and are stored in SqoopOptions object. This object then acts as data transfer object. This object is passed to various phases of processing like preprocessing before running the actual MapReduce, MapReduce phase and even postprocessing phase. This class has a lot of members. The options are parsed and populated in the respective member. Now lets say that a user creates a new user defined tool and this tool has some new options which don’t map to any of the existing members of the SqoopOptions class. Either user can add a new member to SqoopOption class which means users will have to make changes in sqoop and compile it, which mght not be possible always for all users. Other option is to use extraArgs member. This is a string array which contains the options for thirdparty tools which could be passed directly to the third party tool like mysqldump etc. This array string needs parsing every time to understand the parameters. The most elegant way of supporting custom options for user defined tool is customToolOptions map. This is a map member of SqoopOption class. Developer can parse the user defined parameters and populate this map with appropriate key/value pairs. When SqoopOption object is passed to various phases of processing these values will be readily available and parsing is not required for every access. Lets take an example to understand the usage better. Lets say you want to develop a custom tool to merge two hive tables and it will take the following parameters :

None of these options are available in SqoopOption object. Tool Developer can override the applyOptions method and in this method the user options can be parsed and populated in the customToolOptions map. Once that is done, SqoopOption object can be passed throughout program and these values will be available for users.

These option names will be stored as keys and the values passed by users will be stored as values. Lets define these options as static finals :

  public static final String MERGE_KEYS = "merge-keys";
  public static final String HIVE_UPDATES_TABLE = "hive-updates-table";
  public static final String HIVE_UPDATES_TABLE_DB = "hive-updates-database";
  public static final String RETAIN_UPDATES_TBL = "retain-updates-tbl";

A sample applyOptions example which parses the above said options and populates the customToolOptions map is below :

 public void applyOptions(CommandLine in, SqoopOptions out)
    throws InvalidOptionsException {

    if (in.hasOption(VERBOSE_ARG)) {
      LoggingUtils.setDebugLevel();
      log.debug("Enabled debug logging.");
    }

    if (in.hasOption(HELP_ARG)) {
      ToolOptions toolOpts = new ToolOptions();
      configureOptions(toolOpts);
      printHelp(toolOpts);
      throw new InvalidOptionsException("");
    }

    Map<String, String> mergeOptionsMap = new HashMap<String, String>();
    if (in.hasOption(MERGE_KEYS)) {
      mergeOptionsMap.put(MERGE_KEYS, in.getOptionValue(MERGE_KEYS));
    }

    if (in.hasOption(HIVE_UPDATES_TABLE)) {
      mergeOptionsMap.put(HIVE_UPDATES_TABLE,
        in.getOptionValue(HIVE_UPDATES_TABLE));
    }

    if (in.hasOption(HIVE_UPDATES_TABLE_DB)) {
      mergeOptionsMap.put(HIVE_UPDATES_TABLE_DB,
        in.getOptionValue(HIVE_UPDATES_TABLE_DB));
    }

    if (in.hasOption(RETAIN_UPDATES_TBL)) {
      mergeOptionsMap.put(RETAIN_UPDATES_TBL, "");
    }

    if (in.hasOption(HIVE_TABLE_ARG)) {
      out.setHiveTableName(in.getOptionValue(HIVE_TABLE_ARG));
    }

    if (in.hasOption(HIVE_DATABASE_ARG)) {
      out.setHiveDatabaseName(in.getOptionValue(HIVE_DATABASE_ARG));
    }

    if (out.getCustomToolOptions() == null) {
      out.setCustomToolOptions(mergeOptionsMap);
    }
  }

Finally you need to copy your plugin jar to $SQOOP_LIB directory and register the plugin class with sqoop in sqoop-site.xml :

<property>
    <name>sqoop.tool.plugins</name>
    <value>com.expedia.sqoop.tool.HiveMergePlugin</value>
    <description>A comma-delimited list of ToolPlugin implementations
      which are consulted, in order, to register SqoopTool instances which
      allow third-party tools to be used.
    </description>
  </property>