3.3. Sqoop 2 Connector Development¶

3.3.1.1. When do we add a new connector?¶

You add a new connector when you need to extract/read data from a new data source, or load/write data into a new data source that is not supported yet in Sqoop 2. In addition to the connector API, Sqoop 2 also has an submission and execution engine interface. At the moment the only supported engine is MapReduce, but we may support additional engines in the future such as Spark. Since many parallel execution engines are capable of reading/writing data, there may be a question of whether adding support for a new data source should be done through the connector or the execution engine API.

Our guideline are as follows: Connectors should manage all data extract(reading) from and/or load(writing) into a data source. Submission and execution engine together manage the job submission and execution life cycle to read/write data from/to data sources in the most optimal way possible. If you need to support a new data store and details of linking to it and don’t care how the process of reading/writing from/to happens then you are looking to add a connector and you should continue reading the below Connector API details to contribute new connectors to Sqoop 2.

3.3.2.1. From¶

The getFrom method returns From instance which is a Transferable entity that encapsulates the operations needed to read from the data source the connector represents. The built-in GenericJdbcConnector defines From like this.

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);
  ...
}

3.3.2.2. To¶

The getTo method returns TO instance which is a Transferable entity that encapsulates the operations needed to wtite data to the data source the connector represents. The built-in GenericJdbcConnector defines To like this.

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) {
  ...
}

3.3.2.3. Sqoop Connector Identifier : sqoopconnector.properties¶

Every Sqoop 2 connector needs to have a sqoopconnector.properties in the packaged jar to be identified by Sqoop. A typical sqoopconnector.properties for a sqoop2 connector looks like below

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

If the above file does not exist, then Sqoop will not load this jar and thus cannot be registered into Sqoop repository for creating Sqoop jobs

3.3.2.4. Sqoop Connector Build-time Dependencies¶

Sqoop provides the connector-sdk module identified by the package:org.apache.sqoop.connector It provides the public facing apis for the external connectors to extend from. It also provides common utilities that the connectors can utilize for converting data to and from the sqoop intermediate data format

The common-test module identified by the package org.apache.sqoop.common.test provides utilities used related to the built-in connectors such as the JDBC, HDFS, and Kafka connectors that can be used by the external connectors for creating the end-end integration test for sqoop jobs

The test module identified by the package org.apache.sqoop.test provides various minicluster utilites the integration tests can extend from to run

a sqoop job with the given sqoop connector either using it as a FROM or TO data-source

Hence the pom.xml for the sqoop kite connector built using the kite-sdk might look something like below

 <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>

3.3.2.5. Sqoop Connector Build¶

Sqoop 2 supports connectors to package their dependencies into the lib directory inside the connector jar to provide classpath isolation between connectors. Add the following to the pom.xml for the connector:

<plugins>
  <plugin>
    <groupId>org.apache.maven.plugins</groupId>
    <artifactId>maven-assembly-plugin</artifactId>
    <version>${maven-assembly-plugin.version}</version>
    <dependencies>
      <dependency>
        <groupId>org.apache.sqoop</groupId>
        <artifactId>sqoop-assemblies</artifactId>
        <version>${sqoop.version}</version>
      </dependency>
    </dependencies>
    <executions>
      <execution>
        <id>make-assembly</id>
        <phase>package</phase>
        <goals>
          <goal>single</goal>
        </goals>
        <configuration>
          <finalName>${project.artifactId}-${project.version}</finalName>
          <appendAssemblyId>false</appendAssemblyId>
          <descriptorRefs>
            <descriptorRef>sqoop-connector</descriptorRef>
          </descriptorRefs>
        </configuration>
      </execution>
    </executions>
  </plugin>
</plugins>

3.3.3.1. Configurable registration¶

One of the currently supported configurable in Sqoop are the connectors. Sqoop 2 registers definitions of connectors from the file named sqoopconnector.properties which each connector implementation should provide to become available in Sqoop.

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

3.3.3.2. Configurations¶

Implementations of SqoopConnector overrides methods such as getLinkConfigurationClass and getJobConfigurationClass returning configuration class.

@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;
  }
}

Configurations are represented by annotations defined in org.apache.sqoop.model package. Annotations such as ConfigurationClass , ConfigClass , Config and Input are provided for defining configuration objects for each connector.

@ConfigurationClass is a marker annotation for ConfigurationClasses that hold a group or lis of ConfigClasses annotated with the marker @ConfigClass

@ConfigurationClass
public class LinkConfiguration {

  @Config public LinkConfig linkConfig;

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

Each ConfigClass defines the different inputs it exposes for the link and job configs. These inputs are annotated with @Input and the user will be asked to fill in when they create a sqoop job and choose to use this instance of the connector for either the From or To part of the job.

@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;
}

Each ConfigClass and the inputs within the configs annotated with Input can specifiy validators via the @Validator annotation described below.

3.3.3.3. Configs and Inputs¶

As discussed above, Input provides a way to express the type of config parameter exposed. In addition it allows connector developer to add attributes that describe how the input will be used in the sqoop job. Here are the list of the supported attributes

Inputs associated with the link configuration include:

Editable Attribute: Possible values for the Enum InputEditable are USER_ONLY, CONNECTOR_ONLY, ANY. If an input says editable by USER_ONLY, then the connector code during the job run or upgrade cannot update the config input value. Similarly for a CONNECTOR_ONLY, user cannot update its value via the rest api or shell command line.

Overrides Attribute: USER_ONLY input attribute values cannot be overriden by other inputs.

@ConfigurationClass
public class EmptyConfiguration { }

3.3.3.4. Configuration ResourceBundle¶

The config and its corresponding input names, the input field description are represented in the config resource bundle defined per connector.

# 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.

...

Those resources are loaded by getBundle method of the SqoopConnector.

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

3.3.3.5. Validations for Configs and Inputs¶

Validators validate the config objects and the inputs associated with the config objects. For config objects themselves we encourage developers to write custom valdiators for both the link and job config types.

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

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

Sqoop 2 provides a list of standard input validators that can be used by different connectors for the link and job type configuration inputs.

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");
  }
 }
}

The validation logic is executed when users creating the sqoop jobs input values for the link and job configs associated with the From and To instances of the connectors associated with the job.