github/sqoop
5
0
Fork 0
mirror of https://github.com/apache/sqoop.git synced 2025-05-06 17:01:59 +08:00
Code Issues Projects Releases Wiki Activity

SQOOP-2246: Sqoop2: Use jdbcProperties when creating database connection in GenericJDBCExecutor

Browse Source 
(Jarek Jarcec Cecho via Abraham Elmahrek)
This commit is contained in:
Abraham Elmahrek 2015-03-23 16:35:40 -07:00
parent d68737ecfd
commit 598607cde6
12 changed files with 106 additions and 68 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

74
connector/connector-generic-jdbc/src/main/java/org/apache/sqoop/connector/jdbc/GenericJdbcExecutor.java
View File

@ -19,9 +19,11 @@
import org.apache.log4j.Logger;
import org.apache.sqoop.common.SqoopException;
import org.apache.sqoop.connector.jdbc.configuration.LinkConfig;
import org.apache.sqoop.error.code.GenericJdbcConnectorError;
import org.apache.sqoop.schema.Schema;
import org.apache.sqoop.schema.type.Column;
import org.apache.sqoop.utils.ClassUtils;
import org.joda.time.DateTime;
import org.joda.time.LocalDate;
import org.joda.time.LocalTime;
@ -35,25 +37,75 @@
import java.sql.SQLException;
import java.sql.Statement;
import java.sql.Timestamp;
import java.util.Properties;
/**
* Database executor that is based on top of JDBC spec.
*/
public class GenericJdbcExecutor {
private static final Logger LOG =
Logger.getLogger(GenericJdbcExecutor.class);
private static final Logger LOG = Logger.getLogger(GenericJdbcExecutor.class);
/**
* Keys for JDBC properties
*
* We're following JDBC 4 spec:
* http://download.oracle.com/otn-pub/jcp/jdbc-4_1-mrel-spec/jdbc4.1-fr-spec.pdf?AuthParam=1426813649_0155f473b02dbca8bbd417dd061669d7
*/
public static final String JDBC_PROPERTY_USERNAME = "user";
public static final String JDBC_PROPERTY_PASSWORD = "password";
/**
* User configured link with credentials and such
*/
private LinkConfig linkConfig;
/**
* Internal connection object (we'll hold to it)
*/
private Connection connection;
/**
* Prepare statement
*/
private PreparedStatement preparedStatement;
public GenericJdbcExecutor(String driver, String url,
String username, String password) {
public GenericJdbcExecutor(LinkConfig linkConfig) {
assert linkConfig != null;
assert linkConfig.connectionString != null;
// Persist link configuration for future use
this.linkConfig = linkConfig;
// Load/register the JDBC driver to JVM
Class driverClass = ClassUtils.loadClass(linkConfig.jdbcDriver);
if(driverClass == null) {
throw new SqoopException(GenericJdbcConnectorError.GENERIC_JDBC_CONNECTOR_0000, linkConfig.jdbcDriver);
}
// Properties that we will use for the connection
Properties properties = new Properties();
if(linkConfig.jdbcProperties != null) {
properties.putAll(linkConfig.jdbcProperties);
}
// Propagate username and password to the properties
//
// DriverManager have two relevant API for us:
// * getConnection(url, username, password)
// * getConnection(url, properties)
// As we have to use properties, we need to use the later
// method and hence we have to persist the credentials there.
if(linkConfig.username != null) {
properties.put(JDBC_PROPERTY_USERNAME, linkConfig.username);
}
if(linkConfig.password != null) {
properties.put(JDBC_PROPERTY_PASSWORD, linkConfig.password);
}
// Finally create the connection
try {
Class.forName(driver);
connection = DriverManager.getConnection(url, username, password);
} catch (ClassNotFoundException e) {
throw new SqoopException(
GenericJdbcConnectorError.GENERIC_JDBC_CONNECTOR_0000, driver, e);
connection = DriverManager.getConnection(linkConfig.connectionString, properties);
} catch (SQLException e) {
logSQLException(e);
throw new SqoopException(GenericJdbcConnectorError.GENERIC_JDBC_CONNECTOR_0001, e);

9
connector/connector-generic-jdbc/src/main/java/org/apache/sqoop/connector/jdbc/GenericJdbcExtractor.java
View File

@ -41,13 +41,8 @@ public class GenericJdbcExtractor extends Extractor<LinkConfiguration, FromJobCo
private long rowsRead = 0;
@Override
public void extract(ExtractorContext context, LinkConfiguration linkConfig,
FromJobConfiguration fromJobConfig, GenericJdbcPartition partition) {
String driver = linkConfig.linkConfig.jdbcDriver;
String url = linkConfig.linkConfig.connectionString;
String username = linkConfig.linkConfig.username;
String password = linkConfig.linkConfig.password;
GenericJdbcExecutor executor = new GenericJdbcExecutor(driver, url, username, password);
public void extract(ExtractorContext context, LinkConfiguration linkConfig, FromJobConfiguration fromJobConfig, GenericJdbcPartition partition) {
GenericJdbcExecutor executor = new GenericJdbcExecutor(linkConfig.linkConfig);
String query = context.getString(GenericJdbcConnectorConstants.CONNECTOR_JDBC_FROM_DATA_SQL);
String conditions = partition.getConditions();

17
connector/connector-generic-jdbc/src/main/java/org/apache/sqoop/connector/jdbc/GenericJdbcFromInitializer.java
View File

@ -47,7 +47,8 @@ public class GenericJdbcFromInitializer extends Initializer<LinkConfiguration, F
@Override
public void initialize(InitializerContext context, LinkConfiguration linkConfig, FromJobConfiguration fromJobConfig) {
configureJdbcProperties(context.getContext(), linkConfig, fromJobConfig);
executor = new GenericJdbcExecutor(linkConfig.linkConfig);
try {
configurePartitionProperties(context.getContext(), linkConfig, fromJobConfig);
configureTableProperties(context.getContext(), linkConfig, fromJobConfig);
@ -67,7 +68,7 @@ public Set<String> getJars(InitializerContext context, LinkConfiguration linkCon
@Override
public Schema getSchema(InitializerContext context, LinkConfiguration linkConfig, FromJobConfiguration fromJobConfig) {
configureJdbcProperties(context.getContext(), linkConfig, fromJobConfig);
executor = new GenericJdbcExecutor(linkConfig.linkConfig);
String schemaName = fromJobConfig.fromJobConfig.tableName;
if(schemaName == null) {
@ -115,18 +116,6 @@ public Schema getSchema(InitializerContext context, LinkConfiguration linkConfig
}
}
private void configureJdbcProperties(MutableContext context, LinkConfiguration linkConfig, FromJobConfiguration fromJobConfig) {
String driver = linkConfig.linkConfig.jdbcDriver;
String url = linkConfig.linkConfig.connectionString;
String username = linkConfig.linkConfig.username;
String password = linkConfig.linkConfig.password;
assert driver != null;
assert url != null;
executor = new GenericJdbcExecutor(driver, url, username, password);
}
private void configurePartitionProperties(MutableContext context, LinkConfiguration linkConfig, FromJobConfiguration jobConf) throws SQLException {
// Assertions that should be valid (verified via validator)
assert (jobConf.fromJobConfig.tableName != null && jobConf.fromJobConfig.sql == null) ||

6
connector/connector-generic-jdbc/src/main/java/org/apache/sqoop/connector/jdbc/GenericJdbcLoader.java
View File

@ -32,11 +32,7 @@ public class GenericJdbcLoader extends Loader<LinkConfiguration, ToJobConfigurat
@Override
public void load(LoaderContext context, LinkConfiguration linkConfig, ToJobConfiguration toJobConfig) throws Exception{
String driver = linkConfig.linkConfig.jdbcDriver;
String url = linkConfig.linkConfig.connectionString;
String username = linkConfig.linkConfig.username;
String password = linkConfig.linkConfig.password;
GenericJdbcExecutor executor = new GenericJdbcExecutor(driver, url, username, password);
GenericJdbcExecutor executor = new GenericJdbcExecutor(linkConfig.linkConfig);
executor.setAutoCommit(false);
String sql = context.getString(GenericJdbcConnectorConstants.CONNECTOR_JDBC_TO_DATA_SQL);
executor.beginBatch(sql);

9
connector/connector-generic-jdbc/src/main/java/org/apache/sqoop/connector/jdbc/GenericJdbcToDestroyer.java
View File

@ -41,13 +41,8 @@ public void destroy(DestroyerContext context, LinkConfiguration linkConfig, ToJo
}
}
private void moveDataToDestinationTable(LinkConfiguration linkConfig,
boolean success, String stageTableName, String tableName) {
GenericJdbcExecutor executor =
new GenericJdbcExecutor(linkConfig.linkConfig.jdbcDriver,
linkConfig.linkConfig.connectionString,
linkConfig.linkConfig.username,
linkConfig.linkConfig.password);
private void moveDataToDestinationTable(LinkConfiguration linkConfig, boolean success, String stageTableName, String tableName) {
GenericJdbcExecutor executor = new GenericJdbcExecutor(linkConfig.linkConfig);
try {
if(success) {
LOG.info("Job completed, transferring data from stage fromTable to " +

16
connector/connector-generic-jdbc/src/main/java/org/apache/sqoop/connector/jdbc/GenericJdbcToInitializer.java
View File

@ -44,7 +44,7 @@ public class GenericJdbcToInitializer extends Initializer<LinkConfiguration, ToJ
@Override
public void initialize(InitializerContext context, LinkConfiguration linkConfig, ToJobConfiguration toJobConfig) {
configureJdbcProperties(context.getContext(), linkConfig, toJobConfig);
executor = new GenericJdbcExecutor(linkConfig.linkConfig);
try {
configureTableProperties(context.getContext(), linkConfig, toJobConfig);
} finally {
@ -61,7 +61,7 @@ public Set<String> getJars(InitializerContext context, LinkConfiguration linkCon
@Override
public Schema getSchema(InitializerContext context, LinkConfiguration linkConfig, ToJobConfiguration toJobConfig) {
configureJdbcProperties(context.getContext(), linkConfig, toJobConfig);
executor = new GenericJdbcExecutor(linkConfig.linkConfig);
String schemaName = toJobConfig.toJobConfig.tableName;
@ -109,18 +109,6 @@ public Schema getSchema(InitializerContext context, LinkConfiguration linkConfig
}
}
private void configureJdbcProperties(MutableContext context, LinkConfiguration linkConfig, ToJobConfiguration toJobConfig) {
String driver = linkConfig.linkConfig.jdbcDriver;
String url = linkConfig.linkConfig.connectionString;
String username = linkConfig.linkConfig.username;
String password = linkConfig.linkConfig.password;
assert driver != null;
assert url != null;
executor = new GenericJdbcExecutor(driver, url, username, password);
}
private void configureTableProperties(MutableContext context, LinkConfiguration linkConfig, ToJobConfiguration toJobConfig) {
String dataSql;

14
connector/connector-generic-jdbc/src/test/java/org/apache/sqoop/connector/jdbc/GenericJdbcExecutorTest.java
View File

@ -17,6 +17,8 @@
*/
package org.apache.sqoop.connector.jdbc;
import org.apache.sqoop.common.SqoopException;
import org.apache.sqoop.connector.jdbc.configuration.LinkConfig;
import org.testng.annotations.BeforeMethod;
import org.testng.annotations.Test;
@ -33,8 +35,7 @@ public class GenericJdbcExecutorTest {
public GenericJdbcExecutorTest() {
table = getClass().getSimpleName().toUpperCase();
emptyTable = table + "_EMPTY";
executor = new GenericJdbcExecutor(GenericJdbcTestConstants.DRIVER,
GenericJdbcTestConstants.URL, null, null);
executor = new GenericJdbcExecutor(GenericJdbcTestConstants.LINK_CONFIG);
}
@BeforeMethod(alwaysRun = true)
@ -59,6 +60,15 @@ public void setUp() {
}
}
@Test(expectedExceptions = SqoopException.class)
public void testUnknownDriver() {
LinkConfig linkConfig = new LinkConfig();
linkConfig.jdbcDriver = "net.jarcec.driver.MyAwesomeDatabase";
linkConfig.connectionString = "jdbc:awesome:";
new GenericJdbcExecutor(linkConfig);
}
@Test
public void testDeleteTableData() throws Exception {
executor.deleteTableData(table);

17
connector/connector-generic-jdbc/src/test/java/org/apache/sqoop/connector/jdbc/GenericJdbcTestConstants.java
View File

@ -17,9 +17,26 @@
*/
package org.apache.sqoop.connector.jdbc;
import org.apache.sqoop.connector.jdbc.configuration.LinkConfig;
public class GenericJdbcTestConstants {
/**
* Testing Driver
*/
public static final String DRIVER = "org.apache.derby.jdbc.EmbeddedDriver";
/**
* Testing database (in memory derby)
*/
public static final String URL = "jdbc:derby:memory:TESTDB;create=true";
/**
* Test link configuration
*/
public static final LinkConfig LINK_CONFIG = new LinkConfig();
static {
LINK_CONFIG.jdbcDriver = DRIVER;
LINK_CONFIG.connectionString = URL;
}
}

3
connector/connector-generic-jdbc/src/test/java/org/apache/sqoop/connector/jdbc/TestExtractor.java
View File

@ -59,8 +59,7 @@ public TestExtractor() {
@BeforeMethod(alwaysRun = true)
public void setUp() {
executor = new GenericJdbcExecutor(GenericJdbcTestConstants.DRIVER,
GenericJdbcTestConstants.URL, null, null);
executor = new GenericJdbcExecutor(GenericJdbcTestConstants.LINK_CONFIG);
if (!executor.existTable(tableName)) {
executor.executeUpdate("CREATE TABLE "

3
connector/connector-generic-jdbc/src/test/java/org/apache/sqoop/connector/jdbc/TestFromInitializer.java
View File

@ -61,8 +61,7 @@ public TestFromInitializer() {
@BeforeMethod(alwaysRun = true)
public void setUp() {
executor = new GenericJdbcExecutor(GenericJdbcTestConstants.DRIVER,
GenericJdbcTestConstants.URL, null, null);
executor = new GenericJdbcExecutor(GenericJdbcTestConstants.LINK_CONFIG);
String fullTableName = executor.delimitIdentifier(schemaName) + "." + executor.delimitIdentifier(tableName);
if (!executor.existTable(tableName)) {

3
connector/connector-generic-jdbc/src/test/java/org/apache/sqoop/connector/jdbc/TestLoader.java
View File

@ -69,8 +69,7 @@ public TestLoader(int numberOfRows) {
@BeforeMethod(alwaysRun = true)
public void setUp() {
executor = new GenericJdbcExecutor(GenericJdbcTestConstants.DRIVER,
GenericJdbcTestConstants.URL, null, null);
executor = new GenericJdbcExecutor(GenericJdbcTestConstants.LINK_CONFIG);
if (!executor.existTable(tableName)) {
executor.executeUpdate("CREATE TABLE "

3
connector/connector-generic-jdbc/src/test/java/org/apache/sqoop/connector/jdbc/TestToInitializer.java
View File

@ -59,8 +59,7 @@ public TestToInitializer() {
@BeforeMethod(alwaysRun = true)
public void setUp() {
executor = new GenericJdbcExecutor(GenericJdbcTestConstants.DRIVER,
GenericJdbcTestConstants.URL, null, null);
executor = new GenericJdbcExecutor(GenericJdbcTestConstants.LINK_CONFIG);
String fullTableName = executor.delimitIdentifier(schemaName) + "." + executor.delimitIdentifier(tableName);
if (!executor.existTable(tableName)) {