What is a Sqoop Connector?¶

Connectors provide the facility to interact with many data sources and thus can be used as a means to transfer data between them in Sqoop. The connector implementation will provide logic to read from and/or write to a data source that it represents. For instance the ( GenericJdbcConnector ) encapsulates the logic to read from and/or write to jdbc enabled relational data sources. The connector part that enables reading from a data source and transferring this data to internal Sqoop format is called the FROM and the part that enables writng data to a data source by transferring data from Sqoop format is called TO. In order to interact with these data sources, the connector will provide one or many config classes and input fields within it.

Broadly we support two main config types for connectors, link type represented by the enum ConfigType.LINK and job type represented by the enum ConfigType.JOB. Link config represents the properties to physically connect to the data source. Job config represent the properties that are required to invoke reading from and/or writing to particular dataset in the data source it connects to. If a connector supports both reading from and writing to, it will provide the FromJobConfig and ToJobConfig objects. Each of these config objects are custom to each connector and can have one or more inputs associated with each of the Link, FromJob and ToJob config types. Hence we call the connectors as configurables i.e an entity that can provide configs for interacting with the data source it represents. As the connectors evolve over time to support new features in their data sources, the configs and inputs will change as well. Thus the connector API also provides methods for upgrading the config and input names and data related to these data sources across different versions.

The connectors implement logic for various stages of the extract/load process using the connector API described below. While extracting/reading data from the data-source the main stages are Initializer, Partitioner, Extractor and Destroyer. While loading/writitng data to the data source the main stages currently supported are Initializer, Loader and Destroyer. Each stage has its unique set of responsibilities that are explained in detail below. Since connectors understand the internals of the data source they represent, they work in tandem with the sqoop supported execution engines such as MapReduce or Spark (in future) to accomplish this process in a most optimal way.

Connector Implementation¶

The SqoopConnector class defines an API for the connectors that must be implemented by the connector developers. Each Connector must extend SqoopConnector and override the methods shown below.

public abstract String getVersion();
public abstract ResourceBundle getBundle(Locale locale);
public abstract Class getLinkConfigurationClass();
public abstract Class getJobConfigurationClass(Direction direction);
public abstract From getFrom();
public abstract To getTo();
public abstract ConnectorConfigurableUpgrader getConfigurableUpgrader()

Connectors can optionally override the following methods:

public List<Direction> getSupportedDirections();
public Class<? extends IntermediateDataFormat<?>> getIntermediateDataFormat()

The getVersion method returns the current version of the connector It is important to provide a unique identifier every time a connector jar is released externally. In case of the Sqoop built-in connectors, the version refers to the Sqoop build/release version. External connectors can also use the same or similar mechanism to set this version. The version number is critical for the connector upgrade logic used in Sqoop

@Override
 public String getVersion() {
  return VersionInfo.getBuildVersion();
 }

The getFrom method returns From instance which is a Transferable entity that encapsulates the operations needed to read from the data source that the connector represents.

The getTo method returns To instance which is a Transferable entity that encapsulates the operations needed to write to the data source that the connector represents.

Methods such as getBundle , getLinkConfigurationClass , getJobConfigurationClass are related to Configurations

Since a connector represents a data source and it can support one of the two directions, either reading FROM its data source or writing to its data souurce or both, the getSupportedDirections method returns a list of directions that a connector will implement. This should be a subset of the values in the Direction enum we provide:

public List<Direction> getSupportedDirections() {
    return Arrays.asList(new Direction[]{
        Direction.FROM,
        Direction.TO
    });
}

private static final From FROM = new From(
      GenericJdbcFromInitializer.class,
      GenericJdbcPartitioner.class,
      GenericJdbcExtractor.class,
      GenericJdbcFromDestroyer.class);
...

@Override
public From getFrom() {
  return FROM;
}

public abstract void initialize(InitializerContext context, LinkConfiguration linkConfiguration,
    JobConfiguration jobConfiguration);

public List<String> getJars(InitializerContext context, LinkConfiguration linkConfiguration,
    JobConfiguration jobConfiguration){
     return new LinkedList<String>();
    }

public abstract Schema getSchema(InitializerContext context, LinkConfiguration linkConfiguration,
    JobConfiguration jobConfiguration) {
       return new NullSchema();
    }

public abstract List<Partition> getPartitions(PartitionerContext context,
    LinkConfiguration linkConfiguration, FromJobConfiguration jobConfiguration);

public abstract class Partition {
  public abstract void readFields(DataInput in) throws IOException;
  public abstract void write(DataOutput out) throws IOException;
  public abstract String toString();
}

public abstract void extract(ExtractorContext context,
                             LinkConfiguration linkConfiguration,
                             JobConfiguration jobConfiguration,
                             SqoopPartition partition);

context.getDataWriter().writeArrayRecord(array);

while (resultSet.next()) {
  ...
  context.getDataWriter().writeArrayRecord(array);
  ...
}

private static final To TO = new To(
      GenericJdbcToInitializer.class,
      GenericJdbcLoader.class,
      GenericJdbcToDestroyer.class);
...

@Override
public To getTo() {
  return TO;
}

public abstract void load(LoaderContext context,
                          ConnectionConfiguration connectionConfiguration,
                          JobConfiguration jobConfiguration) throws Exception;

while ((array = context.getDataReader().readArrayRecord()) != null) {
  ...
}

# Sqoop Foo Connector Properties
org.apache.sqoop.connector.class = org.apache.sqoop.connector.foo.FooConnector
org.apache.sqoop.connector.name = sqoop-foo-connector

 <dependencies>
  <!-- Sqoop modules -->
  <dependency>
    <groupId>org.apache.sqoop</groupId>
    <artifactId>connector-sdk</artifactId>
  </dependency>

  <!-- Testing specified modules -->
  <dependency>
    <groupId>org.testng</groupId>
    <artifactId>testng</artifactId>
    <scope>test</scope>
  </dependency>
  <dependency>
    <groupId>org.mockito</groupId>
    <artifactId>mockito-all</artifactId>
    <scope>test</scope>
  </dependency>
   <dependency>
     <groupId>org.apache.sqoop</groupId>
     <artifactId>sqoop-common-test</artifactId>
   </dependency>

   <dependency>
     <groupId>org.apache.sqoop</groupId>
     <artifactId>test</artifactId>
   </dependency>
  <!-- Connector required modules -->
  <dependency>
    <groupId>org.kitesdk</groupId>
    <artifactId>kite-data-core</artifactId>
  </dependency>
  ....
</dependencies>

Configurables¶

# Generic JDBC Connector Properties
org.apache.sqoop.connector.class = org.apache.sqoop.connector.jdbc.GenericJdbcConnector
org.apache.sqoop.connector.name = generic-jdbc-connector

@Override
public Class getLinkConfigurationClass() {
  return LinkConfiguration.class;
}

@Override
public Class getJobConfigurationClass(Direction direction) {
  switch (direction) {
    case FROM:
      return FromJobConfiguration.class;
    case TO:
      return ToJobConfiguration.class;
    default:
      return null;
  }
}

@ConfigurationClass
public class LinkConfiguration {

  @Config public LinkConfig linkConfig;

  public LinkConfiguration() {
    linkConfig = new LinkConfig();
  }
}

@ConfigClass(validators = {@Validator(LinkConfig.ConfigValidator.class)})
public class LinkConfig {
  @Input(size = 128, validators = {@Validator(NotEmpty.class), @Validator(ClassAvailable.class)} )
  @Input(size = 128) public String jdbcDriver;
  @Input(size = 128) public String connectionString;
  @Input(size = 40)  public String username;
  @Input(size = 40, sensitive = true) public String password;
  @Input public Map<String, String> jdbcProperties;
}

@ConfigurationClass
public class EmptyConfiguration { }

# jdbc driver
connection.jdbcDriver.label = JDBC Driver Class
connection.jdbcDriver.help = Enter the fully qualified class name of the JDBC \
                   driver that will be used for establishing this connection.

# connect string
connection.connectionString.label = JDBC Connection String
connection.connectionString.help = Enter the value of JDBC connection string to be \
                   used by this connector for creating connections.

...

@Override
public ResourceBundle getBundle(Locale locale) {
  return ResourceBundle.getBundle(
  GenericJdbcConnectorConstants.RESOURCE_BUNDLE_NAME, locale);
}

@Input(size = 128, validators = {@Validator(value = StartsWith.class, strArg = "jdbc:")} )

@Input(size = 255, validators = { @Validator(NotEmpty.class) })

public class NotEmpty extends AbstractValidator<String> {
@Override
public void validate(String instance) {
  if (instance == null || instance.isEmpty()) {
   addMessage(Status.ERROR, "Can't be null nor empty");
  }
 }
}

Loading External Connectors¶

Loading new connector say sqoop-foo-connector to the sqoop2, here are the steps to follow

1. Create a sqoop-foo-connector.jar. Make sure the jar contains the sqoopconnector.properties for it to be picked up by Sqoop
2. Add this jar to the a folder on your installation machine and update the path to this folder in the sqoop.properties located under the server/conf directory under the Sqoop2 for the key org.apache.sqoop.connector.external.loadpath

#
# External connectors load path
# "/path/to/external/connectors/": Add all the connector JARs in the specified folder
#
org.apache.sqoop.connector.external.loadpath=/path/to/connector

3. Start the Sqoop 2 server and while initializing the server this jar should be loaded into the Sqoop 2’s class path and registered into the Sqoop 2 repository

Sqoop 2 MapReduce Job Execution Lifecycle with Connector API¶

Sqoop 2 provides MapReduce utilities such as SqoopMapper and SqoopReducer that aid sqoop job execution.

Note: Any class prefixed with Sqoop is a internal sqoop class provided for MapReduce and is not part of the conenector API. These internal classes work with the custom implementations of Extractor, Partitioner in the From instance and Loader in the To instance of the connector.

When reading from a data source, the Extractor provided by the From instance of the connector extracts data from a corresponding data source it represents and the Loader, provided by the TO instance of the connector, loads data into the data source it represents.

The diagram below describes the initialization phase of a job. SqoopInputFormat create splits using Partitioner.

    ,----------------.          ,-----------.
    |SqoopInputFormat|          |Partitioner|
    `-------+--------'          `-----+-----'
 getSplits  |                         |
----------->|                         |
            |      getPartitions      |
            |------------------------>|
            |                         |         ,---------.
            |                         |-------> |Partition|
            |                         |         `----+----'
            |<- - - - - - - - - - - - |              |
            |                         |              |          ,----------.
            |-------------------------------------------------->|SqoopSplit|
            |                         |              |          `----+-----'

The diagram below describes the map phase of a job. SqoopMapper invokes From connector’s extractor’s extract method.

    ,-----------.
    |SqoopMapper|
    `-----+-----'
   run    |
--------->|                                   ,------------------.
          |---------------------------------->|SqoopMapDataWriter|
          |                                   `------+-----------'
          |                ,---------.               |
          |--------------> |Extractor|               |
          |                `----+----'               |
          |      extract        |                    |
          |-------------------->|                    |
          |                     |                    |
         read from Data Source  |                    |
<-------------------------------|      write*        |
          |                     |------------------->|
          |                     |                    |           ,-------------.
          |                     |                    |---------->|SqoopWritable|
          |                     |                    |           `----+--------'
          |                     |                    |                |
          |                     |                    |                |  context.write(writable, ..)
          |                     |                    |                |---------------------------->

The diagram below decribes the reduce phase of a job. OutputFormat invokes To connector’s loader’s load method (via SqoopOutputFormatLoadExecutor ).

  ,------------.  ,---------------------.
  |SqoopReducer|  |SqoopNullOutputFormat|
  `---+--------'  `----------+----------'
      |                 |   ,-----------------------------.
      |                 |-> |SqoopOutputFormatLoadExecutor|
      |                 |   `--------------+--------------'              |
      |                 |                  |                             |
      |                 |                  |   ,-----------------.   ,-------------.
      |                 |                  |-> |SqoopRecordWriter|-->|SqoopWritable|
    getRecordWriter     |                  |   `--------+--------'   `---+---------'
----------------------->| getRecordWriter  |            |                |
      |                 |----------------->|            |                |     ,--------------.
      |                 |                  |---------------------------------->|ConsumerThread|
      |                 |                  |            |                |     `------+-------'
      |                 |<- - - - - - - - -|            |                |            |    ,------.
<- - - - - - - - - - - -|                  |            |                |            |--->|Loader|
      |                 |                  |            |                |            |    `--+---'
      |                 |                  |            |                |            |       |
      |                 |                  |            |                |            | load  |
 run  |                 |                  |            |                |            |------>|
----->|                 |     write        |            |                |            |       |
      |------------------------------------------------>| setContent     |            | read* |
      |                 |                  |            |--------------->| getContent |<------|
      |                 |                  |            |                |<-----------|       |
      |                 |                  |            |                |            | - - ->|
      |                 |                  |            |                |            |       | write into Data Source
      |                 |                  |            |                |            |       |----------------------->

More details can be found in Sqoop MR Execution Engine