| /** |
| * Licensed to the Apache Software Foundation (ASF) under one |
| * or more contributor license agreements. See the NOTICE file |
| * distributed with this work for additional information |
| * regarding copyright ownership. The ASF licenses this file |
| * to you under the Apache License, Version 2.0 (the |
| * "License"); you may not use this file except in compliance |
| * with the License. You may obtain a copy of the License at |
| * |
| * http://www.apache.org/licenses/LICENSE-2.0 |
| * |
| * Unless required by applicable law or agreed to in writing, software |
| * distributed under the License is distributed on an "AS IS" BASIS, |
| * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. |
| * See the License for the specific language governing permissions and |
| * limitations under the License. |
| */ |
| |
| package org.apache.sqoop.tool; |
| |
| import java.io.IOException; |
| import java.sql.Connection; |
| import java.sql.ResultSet; |
| import java.sql.ResultSetMetaData; |
| import java.sql.SQLException; |
| import java.sql.Statement; |
| import java.sql.Types; |
| import java.util.List; |
| import java.util.Map; |
| |
| import org.apache.commons.cli.CommandLine; |
| import org.apache.commons.cli.OptionBuilder; |
| import org.apache.commons.logging.Log; |
| import org.apache.commons.logging.LogFactory; |
| import org.apache.hadoop.conf.Configuration; |
| import org.apache.hadoop.fs.FileSystem; |
| import org.apache.hadoop.fs.Path; |
| import org.apache.hadoop.util.StringUtils; |
| import org.apache.sqoop.avro.AvroSchemaMismatchException; |
| |
| import org.apache.sqoop.SqoopOptions; |
| import org.apache.sqoop.SqoopOptions.InvalidOptionsException; |
| import org.apache.sqoop.cli.RelatedOptions; |
| import org.apache.sqoop.cli.ToolOptions; |
| import org.apache.sqoop.hive.HiveClient; |
| import org.apache.sqoop.hive.HiveClientFactory; |
| import org.apache.sqoop.manager.ImportJobContext; |
| import org.apache.sqoop.mapreduce.MergeJob; |
| import org.apache.sqoop.metastore.JobData; |
| import org.apache.sqoop.metastore.JobStorage; |
| import org.apache.sqoop.metastore.JobStorageFactory; |
| import org.apache.sqoop.orm.ClassWriter; |
| import org.apache.sqoop.orm.TableClassName; |
| import org.apache.sqoop.util.AppendUtils; |
| import org.apache.sqoop.util.ClassLoaderStack; |
| import org.apache.sqoop.util.ImportException; |
| |
| import static org.apache.sqoop.manager.SupportedManagers.MYSQL; |
| |
| /** |
| * Tool that performs database imports to HDFS. |
| */ |
| public class ImportTool extends BaseSqoopTool { |
| |
| public static final Log LOG = LogFactory.getLog(ImportTool.class.getName()); |
| |
| private static final String IMPORT_FAILED_ERROR_MSG = "Import failed: "; |
| |
| private CodeGenTool codeGenerator; |
| |
| // true if this is an all-tables import. Set by a subclass which |
| // overrides the run() method of this tool (which can only do |
| // a single table). |
| private boolean allTables; |
| |
| // store check column type for incremental option |
| private int checkColumnType; |
| |
| // Set classloader for local job runner |
| private ClassLoader prevClassLoader = null; |
| |
| private final HiveClientFactory hiveClientFactory; |
| |
| public ImportTool() { |
| this("import", false); |
| } |
| |
| public ImportTool(String toolName, boolean allTables) { |
| this(toolName, new CodeGenTool(), allTables, new HiveClientFactory()); |
| } |
| |
| public ImportTool(String toolName, CodeGenTool codeGenerator, boolean allTables, HiveClientFactory hiveClientFactory) { |
| super(toolName); |
| this.codeGenerator = codeGenerator; |
| this.allTables = allTables; |
| this.hiveClientFactory = hiveClientFactory; |
| } |
| |
| @Override |
| protected boolean init(SqoopOptions sqoopOpts) { |
| boolean ret = super.init(sqoopOpts); |
| codeGenerator.setManager(manager); |
| return ret; |
| } |
| |
| /** |
| * @return a list of jar files generated as part of this import process |
| */ |
| public List<String> getGeneratedJarFiles() { |
| return this.codeGenerator.getGeneratedJarFiles(); |
| } |
| |
| /** |
| * If jars must be loaded into the local environment, do so here. |
| */ |
| private void loadJars(Configuration conf, String ormJarFile, |
| String tableClassName) throws IOException { |
| |
| boolean isLocal = "local".equals(conf.get("mapreduce.jobtracker.address")) |
| || "local".equals(conf.get("mapred.job.tracker")); |
| if (isLocal) { |
| // If we're using the LocalJobRunner, then instead of using the compiled |
| // jar file as the job source, we're running in the current thread. Push |
| // on another classloader that loads from that jar in addition to |
| // everything currently on the classpath. |
| this.prevClassLoader = ClassLoaderStack.addJarFile(ormJarFile, |
| tableClassName); |
| } |
| } |
| |
| /** |
| * If any classloader was invoked by loadJars, free it here. |
| */ |
| private void unloadJars() { |
| if (null != this.prevClassLoader) { |
| // unload the special classloader for this jar. |
| ClassLoaderStack.setCurrentClassLoader(this.prevClassLoader); |
| } |
| } |
| |
| /** |
| * @return true if the supplied options specify an incremental import. |
| */ |
| private boolean isIncremental(SqoopOptions options) { |
| return !options.getIncrementalMode().equals( |
| SqoopOptions.IncrementalMode.None); |
| } |
| |
| /** |
| * If this is an incremental import, then we should save the |
| * user's state back to the metastore (if this job was run |
| * from the metastore). Otherwise, log to the user what data |
| * they need to supply next time. |
| */ |
| private void saveIncrementalState(SqoopOptions options) |
| throws IOException { |
| if (!isIncremental(options)) { |
| return; |
| } |
| |
| Map<String, String> descriptor = options.getStorageDescriptor(); |
| String jobName = options.getJobName(); |
| |
| if (null != jobName && null != descriptor) { |
| // Actually save it back to the metastore. |
| LOG.info("Saving incremental import state to the metastore"); |
| JobStorageFactory ssf = new JobStorageFactory(options.getConf()); |
| JobStorage storage = ssf.getJobStorage(descriptor); |
| storage.open(descriptor); |
| try { |
| // Save the 'parent' SqoopOptions; this does not contain the mutations |
| // to the SqoopOptions state that occurred over the course of this |
| // execution, except for the one we specifically want to memorize: |
| // the latest value of the check column. |
| JobData data = new JobData(options.getParent(), this); |
| storage.update(jobName, data); |
| LOG.info("Updated data for job: " + jobName); |
| } finally { |
| storage.close(); |
| } |
| } else { |
| // If there wasn't a parent SqoopOptions, then the incremental |
| // state data was stored in the current SqoopOptions. |
| LOG.info("Incremental import complete! To run another incremental " |
| + "import of all data following this import, supply the " |
| + "following arguments:"); |
| SqoopOptions.IncrementalMode incrementalMode = |
| options.getIncrementalMode(); |
| switch (incrementalMode) { |
| case AppendRows: |
| LOG.info(" --incremental append"); |
| break; |
| case DateLastModified: |
| LOG.info(" --incremental lastmodified"); |
| break; |
| default: |
| LOG.warn("Undefined incremental mode: " + incrementalMode); |
| break; |
| } |
| LOG.info(" --check-column " + options.getIncrementalTestColumn()); |
| LOG.info(" --last-value " + options.getIncrementalLastValue()); |
| LOG.info("(Consider saving this with 'sqoop job --create')"); |
| } |
| } |
| |
| /** |
| * Return the max value in the incremental-import test column. This |
| * value must be numeric. |
| */ |
| private Object getMaxColumnId(SqoopOptions options) throws SQLException { |
| StringBuilder sb = new StringBuilder(); |
| String query; |
| |
| sb.append("SELECT MAX("); |
| sb.append(manager.escapeColName(options.getIncrementalTestColumn())); |
| sb.append(") FROM "); |
| |
| if (options.getTableName() != null) { |
| // Table import |
| sb.append(manager.escapeTableName(options.getTableName())); |
| |
| String where = options.getWhereClause(); |
| if (null != where) { |
| sb.append(" WHERE "); |
| sb.append(where); |
| } |
| query = sb.toString(); |
| } else { |
| // Free form table based import |
| sb.append("("); |
| sb.append(options.getSqlQuery()); |
| sb.append(") sqoop_import_query_alias"); |
| |
| query = sb.toString().replaceAll("\\$CONDITIONS", "(1 = 1)"); |
| } |
| |
| Connection conn = manager.getConnection(); |
| Statement s = null; |
| ResultSet rs = null; |
| try { |
| LOG.info("Maximal id query for free form incremental import: " + query); |
| s = conn.createStatement(); |
| rs = s.executeQuery(query); |
| if (!rs.next()) { |
| // This probably means the table is empty. |
| LOG.warn("Unexpected: empty results for max value query?"); |
| return null; |
| } |
| |
| ResultSetMetaData rsmd = rs.getMetaData(); |
| checkColumnType = rsmd.getColumnType(1); |
| if (checkColumnType == Types.TIMESTAMP) { |
| return rs.getTimestamp(1); |
| } else if (checkColumnType == Types.DATE) { |
| return rs.getDate(1); |
| } else if (checkColumnType == Types.TIME) { |
| return rs.getTime(1); |
| } else { |
| return rs.getObject(1); |
| } |
| } finally { |
| try { |
| if (null != rs) { |
| rs.close(); |
| } |
| } catch (SQLException sqlE) { |
| LOG.warn("SQL Exception closing resultset: " + sqlE); |
| } |
| |
| try { |
| if (null != s) { |
| s.close(); |
| } |
| } catch (SQLException sqlE) { |
| LOG.warn("SQL Exception closing statement: " + sqlE); |
| } |
| } |
| } |
| |
| /** |
| * Determine if a column is date/time. |
| * @return true if column type is TIMESTAMP, DATE, or TIME. |
| */ |
| private boolean isDateTimeColumn(int columnType) { |
| return (columnType == Types.TIMESTAMP) |
| || (columnType == Types.DATE) |
| || (columnType == Types.TIME); |
| } |
| |
| /** |
| * Initialize the constraints which set the incremental import range. |
| * @return false if an import is not necessary, because the dataset has not |
| * changed. |
| */ |
| private boolean initIncrementalConstraints(SqoopOptions options, |
| ImportJobContext context) throws ImportException, IOException { |
| |
| // If this is an incremental import, determine the constraints |
| // to inject in the WHERE clause or $CONDITIONS for a query. |
| // Also modify the 'last value' field of the SqoopOptions to |
| // specify the current job start time / start row. |
| |
| if (!isIncremental(options)) { |
| return true; |
| } |
| |
| SqoopOptions.IncrementalMode incrementalMode = options.getIncrementalMode(); |
| String nextIncrementalValue = null; |
| |
| Object nextVal; |
| switch (incrementalMode) { |
| case AppendRows: |
| try { |
| nextVal = getMaxColumnId(options); |
| if (isDateTimeColumn(checkColumnType)) { |
| nextIncrementalValue = (nextVal == null) ? null |
| : manager.datetimeToQueryString(nextVal.toString(), |
| checkColumnType); |
| } else if (manager.isCharColumn(checkColumnType)) { |
| throw new ImportException("Character column " |
| + "(" + options.getIncrementalTestColumn() + ") can not be used " |
| + "to determine which rows to incrementally import."); |
| } else { |
| nextIncrementalValue = (nextVal == null) ? null : nextVal.toString(); |
| } |
| } catch (SQLException sqlE) { |
| throw new IOException(sqlE); |
| } |
| break; |
| case DateLastModified: |
| if (options.getMergeKeyCol() == null && !options.isAppendMode()) { |
| Path outputPath = getOutputPath(options, context.getTableName(), false); |
| FileSystem fs = outputPath.getFileSystem(options.getConf()); |
| if (fs.exists(outputPath)) { |
| throw new ImportException("--" + MERGE_KEY_ARG + " or " + "--" + APPEND_ARG |
| + " is required when using --" + this.INCREMENT_TYPE_ARG |
| + " lastmodified and the output directory exists."); |
| } |
| } |
| checkColumnType = manager.getColumnTypes(options.getTableName(), |
| options.getSqlQuery()).get(options.getIncrementalTestColumn()); |
| nextVal = manager.getCurrentDbTimestamp(); |
| if (null == nextVal) { |
| throw new IOException("Could not get current time from database"); |
| } |
| nextIncrementalValue = manager.datetimeToQueryString(nextVal.toString(), |
| checkColumnType); |
| break; |
| default: |
| throw new ImportException("Undefined incremental import type: " |
| + incrementalMode); |
| } |
| |
| // Build the WHERE clause components that are used to import |
| // only this incremental section. |
| StringBuilder sb = new StringBuilder(); |
| String prevEndpoint = options.getIncrementalLastValue(); |
| |
| if (isDateTimeColumn(checkColumnType) && null != prevEndpoint |
| && !prevEndpoint.startsWith("\'") && !prevEndpoint.endsWith("\'")) { |
| // Incremental imports based on date/time should be 'quoted' in |
| // ANSI SQL. If the user didn't specify single-quotes, put them |
| // around, here. |
| prevEndpoint = manager.datetimeToQueryString(prevEndpoint, |
| checkColumnType); |
| } |
| |
| String checkColName = manager.escapeColName( |
| options.getIncrementalTestColumn()); |
| LOG.info("Incremental import based on column " + checkColName); |
| if (null != prevEndpoint) { |
| if (prevEndpoint.equals(nextIncrementalValue)) { |
| LOG.info("No new rows detected since last import."); |
| return false; |
| } |
| LOG.info("Lower bound value: " + prevEndpoint); |
| sb.append(checkColName); |
| switch (incrementalMode) { |
| case AppendRows: |
| sb.append(" > "); |
| break; |
| case DateLastModified: |
| sb.append(" >= "); |
| break; |
| default: |
| throw new ImportException("Undefined comparison"); |
| } |
| sb.append(prevEndpoint); |
| sb.append(" AND "); |
| } |
| |
| if (null != nextIncrementalValue) { |
| sb.append(checkColName); |
| switch (incrementalMode) { |
| case AppendRows: |
| sb.append(" <= "); |
| break; |
| case DateLastModified: |
| sb.append(" < "); |
| break; |
| default: |
| throw new ImportException("Undefined comparison"); |
| } |
| sb.append(nextIncrementalValue); |
| } else { |
| sb.append(checkColName); |
| sb.append(" IS NULL "); |
| } |
| |
| LOG.info("Upper bound value: " + nextIncrementalValue); |
| |
| if (options.getTableName() != null) { |
| // Table based import |
| String prevWhereClause = options.getWhereClause(); |
| if (null != prevWhereClause) { |
| sb.append(" AND ("); |
| sb.append(prevWhereClause); |
| sb.append(")"); |
| } |
| |
| String newConstraints = sb.toString(); |
| options.setWhereClause(newConstraints); |
| } else { |
| // Incremental based import |
| sb.append(" AND $CONDITIONS"); |
| String newQuery = options.getSqlQuery().replace( |
| "$CONDITIONS", sb.toString()); |
| options.setSqlQuery(newQuery); |
| } |
| // Save this state for next time. |
| SqoopOptions recordOptions = options.getParent(); |
| if (null == recordOptions) { |
| recordOptions = options; |
| } |
| recordOptions.setIncrementalLastValue( |
| (nextVal == null) ? null : nextVal.toString()); |
| |
| return true; |
| } |
| |
| /** |
| * Merge HDFS output directories |
| */ |
| protected void lastModifiedMerge(SqoopOptions options, ImportJobContext context) throws IOException { |
| if (context.getDestination() == null) { |
| return; |
| } |
| |
| Path userDestDir = getOutputPath(options, context.getTableName(), false); |
| FileSystem fs = userDestDir.getFileSystem(options.getConf()); |
| if (fs.exists(context.getDestination())) { |
| LOG.info("Final destination exists, will run merge job."); |
| if (fs.exists(userDestDir)) { |
| String tableClassName = null; |
| if (!context.getConnManager().isORMFacilitySelfManaged()) { |
| tableClassName = |
| new TableClassName(options).getClassForTable(context.getTableName()); |
| } |
| Path destDir = getOutputPath(options, context.getTableName()); |
| options.setExistingJarName(context.getJarFile()); |
| options.setClassName(tableClassName); |
| options.setMergeOldPath(userDestDir.toString()); |
| options.setMergeNewPath(context.getDestination().toString()); |
| // Merge to temporary directory so that original directory remains intact. |
| options.setTargetDir(destDir.toString()); |
| |
| // Local job tracker needs jars in the classpath. |
| if (options.getFileLayout() == SqoopOptions.FileLayout.ParquetFile) { |
| loadJars(options.getConf(), context.getJarFile(), ClassWriter.toJavaIdentifier("codegen_" + |
| context.getTableName())); |
| } else { |
| loadJars(options.getConf(), context.getJarFile(), context.getTableName()); |
| } |
| |
| MergeJob mergeJob = new MergeJob(options); |
| if (mergeJob.runMergeJob()) { |
| // Rename destination directory to proper location. |
| Path tmpDir = getOutputPath(options, context.getTableName()); |
| fs.rename(userDestDir, tmpDir); |
| fs.rename(destDir, userDestDir); |
| fs.delete(tmpDir, true); |
| } else { |
| LOG.error("Merge MapReduce job failed!"); |
| } |
| |
| unloadJars(); |
| } else { |
| // Create parent directory(ies), otherwise fs.rename would fail |
| if(!fs.exists(userDestDir.getParent())) { |
| fs.mkdirs(userDestDir.getParent()); |
| } |
| |
| // And finally move the data |
| LOG.info("Moving data from temporary directory " + context.getDestination() + " to final destination " + userDestDir); |
| if(!fs.rename(context.getDestination(), userDestDir)) { |
| throw new RuntimeException("Couldn't move data from temporary directory " + context.getDestination() + " to final destination " + userDestDir); |
| } |
| } |
| } |
| } |
| |
| /** |
| * Import a table or query. |
| * @return true if an import was performed, false otherwise. |
| */ |
| protected boolean importTable(SqoopOptions options) throws IOException, ImportException { |
| String jarFile = null; |
| |
| // Generate the ORM code for the tables. |
| jarFile = codeGenerator.generateORM(options, options.getTableName()); |
| |
| Path outputPath = getOutputPath(options, options.getTableName()); |
| |
| // Do the actual import. |
| ImportJobContext context = new ImportJobContext(options.getTableName(), jarFile, |
| options, outputPath); |
| |
| // If we're doing an incremental import, set up the |
| // filtering conditions used to get the latest records. |
| if (!initIncrementalConstraints(options, context)) { |
| return false; |
| } |
| |
| if (options.isDeleteMode()) { |
| deleteTargetDir(context); |
| } |
| |
| if (null != options.getTableName()) { |
| manager.importTable(context); |
| } else { |
| manager.importQuery(context); |
| } |
| |
| if (options.isAppendMode()) { |
| AppendUtils app = new AppendUtils(context); |
| app.append(); |
| } else if (options.getIncrementalMode() == SqoopOptions.IncrementalMode.DateLastModified) { |
| lastModifiedMerge(options, context); |
| } |
| |
| // If the user wants this table to be in Hive, perform that post-load. |
| if (options.doHiveImport()) { |
| // For Parquet file, the import action will create hive table directly via |
| // kite. So there is no need to do hive import as a post step again. |
| if (options.getFileLayout() != SqoopOptions.FileLayout.ParquetFile) { |
| HiveClient hiveClient = hiveClientFactory.createHiveClient(options, manager); |
| hiveClient.importTable(); |
| } |
| } |
| |
| saveIncrementalState(options); |
| |
| return true; |
| } |
| |
| private void deleteTargetDir(ImportJobContext context) throws IOException { |
| |
| SqoopOptions options = context.getOptions(); |
| Path destDir = context.getDestination(); |
| FileSystem fs = destDir.getFileSystem(options.getConf()); |
| |
| if (fs.exists(destDir)) { |
| fs.delete(destDir, true); |
| LOG.info("Destination directory " + destDir + " deleted."); |
| return; |
| } else { |
| LOG.info("Destination directory " + destDir + " is not present, " |
| + "hence not deleting."); |
| } |
| } |
| |
| /** |
| * @return the output path for the imported files; |
| * in append mode this will point to a temporary folder. |
| * if importing to hbase, this may return null. |
| */ |
| private Path getOutputPath(SqoopOptions options, String tableName) { |
| return getOutputPath(options, tableName, options.isAppendMode() |
| || options.getIncrementalMode().equals(SqoopOptions.IncrementalMode.DateLastModified)); |
| } |
| |
| /** |
| * @return the output path for the imported files; |
| * if importing to hbase, this may return null. |
| */ |
| private Path getOutputPath(SqoopOptions options, String tableName, boolean temp) { |
| // Get output directory |
| String hdfsWarehouseDir = options.getWarehouseDir(); |
| String hdfsTargetDir = options.getTargetDir(); |
| Path outputPath = null; |
| if (temp) { |
| // Use temporary path, later removed when appending |
| String salt = tableName; |
| if(salt == null && options.getSqlQuery() != null) { |
| salt = Integer.toHexString(options.getSqlQuery().hashCode()); |
| } |
| outputPath = AppendUtils.getTempAppendDir(salt, options); |
| LOG.debug("Using temporary folder: " + outputPath.getName()); |
| } else { |
| // Try in this order: target-dir or warehouse-dir |
| if (hdfsTargetDir != null) { |
| outputPath = new Path(hdfsTargetDir); |
| } else if (hdfsWarehouseDir != null) { |
| outputPath = new Path(hdfsWarehouseDir, tableName); |
| } else if (null != tableName) { |
| outputPath = new Path(tableName); |
| } |
| } |
| |
| return outputPath; |
| } |
| |
| @Override |
| /** {@inheritDoc} */ |
| public int run(SqoopOptions options) { |
| if (allTables) { |
| // We got into this method, but we should be in a subclass. |
| // (This method only handles a single table) |
| // This should not be reached, but for sanity's sake, test here. |
| LOG.error("ImportTool.run() can only handle a single table."); |
| return 1; |
| } |
| |
| if (!init(options)) { |
| return 1; |
| } |
| |
| codeGenerator.setManager(manager); |
| |
| try { |
| // Import a single table (or query) the user specified. |
| importTable(options); |
| } catch (IllegalArgumentException iea) { |
| LOG.error(IMPORT_FAILED_ERROR_MSG + iea.getMessage()); |
| rethrowIfRequired(options, iea); |
| return 1; |
| } catch (IOException ioe) { |
| LOG.error(IMPORT_FAILED_ERROR_MSG + StringUtils.stringifyException(ioe)); |
| rethrowIfRequired(options, ioe); |
| return 1; |
| } catch (ImportException ie) { |
| LOG.error(IMPORT_FAILED_ERROR_MSG + ie.toString()); |
| rethrowIfRequired(options, ie); |
| return 1; |
| } catch (AvroSchemaMismatchException e) { |
| LOG.error(IMPORT_FAILED_ERROR_MSG, e); |
| rethrowIfRequired(options, e); |
| return 1; |
| } finally { |
| destroy(options); |
| } |
| |
| return 0; |
| } |
| |
| /** |
| * Construct the set of options that control imports, either of one |
| * table or a batch of tables. |
| * @return the RelatedOptions that can be used to parse the import |
| * arguments. |
| */ |
| @SuppressWarnings("static-access") |
| protected RelatedOptions getImportOptions() { |
| // Imports |
| RelatedOptions importOpts = new RelatedOptions("Import control arguments"); |
| |
| importOpts.addOption(OptionBuilder |
| .withDescription("Use direct import fast path") |
| .withLongOpt(DIRECT_ARG) |
| .create()); |
| |
| if (!allTables) { |
| importOpts.addOption(OptionBuilder.withArgName("table-name") |
| .hasArg().withDescription("Table to read") |
| .withLongOpt(TABLE_ARG) |
| .create()); |
| importOpts.addOption(OptionBuilder.withArgName("col,col,col...") |
| .hasArg().withDescription("Columns to import from table") |
| .withLongOpt(COLUMNS_ARG) |
| .create()); |
| importOpts.addOption(OptionBuilder.withArgName("column-name") |
| .hasArg() |
| .withDescription("Column of the table used to split work units") |
| .withLongOpt(SPLIT_BY_ARG) |
| .create()); |
| importOpts |
| .addOption(OptionBuilder |
| .withArgName("size") |
| .hasArg() |
| .withDescription( |
| "Upper Limit of rows per split for split columns of Date/Time/Timestamp and integer types. For date or timestamp fields it is calculated in seconds. split-limit should be greater than 0") |
| .withLongOpt(SPLIT_LIMIT_ARG) |
| .create()); |
| importOpts.addOption(OptionBuilder.withArgName("where clause") |
| .hasArg().withDescription("WHERE clause to use during import") |
| .withLongOpt(WHERE_ARG) |
| .create()); |
| importOpts.addOption(OptionBuilder |
| .withDescription("Imports data in append mode") |
| .withLongOpt(APPEND_ARG) |
| .create()); |
| importOpts.addOption(OptionBuilder |
| .withDescription("Imports data in delete mode") |
| .withLongOpt(DELETE_ARG) |
| .create()); |
| importOpts.addOption(OptionBuilder.withArgName("dir") |
| .hasArg().withDescription("HDFS plain table destination") |
| .withLongOpt(TARGET_DIR_ARG) |
| .create()); |
| importOpts.addOption(OptionBuilder.withArgName("statement") |
| .hasArg() |
| .withDescription("Import results of SQL 'statement'") |
| .withLongOpt(SQL_QUERY_ARG) |
| .create(SQL_QUERY_SHORT_ARG)); |
| importOpts.addOption(OptionBuilder.withArgName("statement") |
| .hasArg() |
| .withDescription("Set boundary query for retrieving max and min" |
| + " value of the primary key") |
| .withLongOpt(SQL_QUERY_BOUNDARY) |
| .create()); |
| importOpts.addOption(OptionBuilder.withArgName("column") |
| .hasArg().withDescription("Key column to use to join results") |
| .withLongOpt(MERGE_KEY_ARG) |
| .create()); |
| |
| addValidationOpts(importOpts); |
| } |
| |
| importOpts.addOption(OptionBuilder.withArgName("dir") |
| .hasArg().withDescription("HDFS parent for table destination") |
| .withLongOpt(WAREHOUSE_DIR_ARG) |
| .create()); |
| importOpts.addOption(OptionBuilder |
| .withDescription("Imports data to SequenceFiles") |
| .withLongOpt(FMT_SEQUENCEFILE_ARG) |
| .create()); |
| importOpts.addOption(OptionBuilder |
| .withDescription("Imports data as plain text (default)") |
| .withLongOpt(FMT_TEXTFILE_ARG) |
| .create()); |
| importOpts.addOption(OptionBuilder |
| .withDescription("Imports data to Avro data files") |
| .withLongOpt(FMT_AVRODATAFILE_ARG) |
| .create()); |
| importOpts.addOption(OptionBuilder |
| .withDescription("Imports data to Parquet files") |
| .withLongOpt(BaseSqoopTool.FMT_PARQUETFILE_ARG) |
| .create()); |
| importOpts.addOption(OptionBuilder.withArgName("n") |
| .hasArg().withDescription("Use 'n' map tasks to import in parallel") |
| .withLongOpt(NUM_MAPPERS_ARG) |
| .create(NUM_MAPPERS_SHORT_ARG)); |
| importOpts.addOption(OptionBuilder.withArgName("name") |
| .hasArg().withDescription("Set name for generated mapreduce job") |
| .withLongOpt(MAPREDUCE_JOB_NAME) |
| .create()); |
| importOpts.addOption(OptionBuilder |
| .withDescription("Enable compression") |
| .withLongOpt(COMPRESS_ARG) |
| .create(COMPRESS_SHORT_ARG)); |
| importOpts.addOption(OptionBuilder.withArgName("codec") |
| .hasArg() |
| .withDescription("Compression codec to use for import") |
| .withLongOpt(COMPRESSION_CODEC_ARG) |
| .create()); |
| importOpts.addOption(OptionBuilder.withArgName("n") |
| .hasArg() |
| .withDescription("Split the input stream every 'n' bytes " |
| + "when importing in direct mode") |
| .withLongOpt(DIRECT_SPLIT_SIZE_ARG) |
| .create()); |
| importOpts.addOption(OptionBuilder.withArgName("n") |
| .hasArg() |
| .withDescription("Set the maximum size for an inline LOB") |
| .withLongOpt(INLINE_LOB_LIMIT_ARG) |
| .create()); |
| importOpts.addOption(OptionBuilder.withArgName("n") |
| .hasArg() |
| .withDescription("Set number 'n' of rows to fetch from the " |
| + "database when more rows are needed") |
| .withLongOpt(FETCH_SIZE_ARG) |
| .create()); |
| importOpts.addOption(OptionBuilder.withArgName("reset-mappers") |
| .withDescription("Reset the number of mappers to one mapper if no split key available") |
| .withLongOpt(AUTORESET_TO_ONE_MAPPER) |
| .create()); |
| return importOpts; |
| } |
| |
| /** |
| * Return options for incremental import. |
| */ |
| protected RelatedOptions getIncrementalOptions() { |
| RelatedOptions incrementalOpts = |
| new RelatedOptions("Incremental import arguments"); |
| |
| incrementalOpts.addOption(OptionBuilder.withArgName("import-type") |
| .hasArg() |
| .withDescription( |
| "Define an incremental import of type 'append' or 'lastmodified'") |
| .withLongOpt(INCREMENT_TYPE_ARG) |
| .create()); |
| incrementalOpts.addOption(OptionBuilder.withArgName("column") |
| .hasArg() |
| .withDescription("Source column to check for incremental change") |
| .withLongOpt(INCREMENT_COL_ARG) |
| .create()); |
| incrementalOpts.addOption(OptionBuilder.withArgName("value") |
| .hasArg() |
| .withDescription("Last imported value in the incremental check column") |
| .withLongOpt(INCREMENT_LAST_VAL_ARG) |
| .create()); |
| |
| return incrementalOpts; |
| } |
| |
| @Override |
| /** Configure the command-line arguments we expect to receive */ |
| public void configureOptions(ToolOptions toolOptions) { |
| |
| toolOptions.addUniqueOptions(getCommonOptions()); |
| toolOptions.addUniqueOptions(getImportOptions()); |
| if (!allTables) { |
| toolOptions.addUniqueOptions(getIncrementalOptions()); |
| } |
| toolOptions.addUniqueOptions(getOutputFormatOptions()); |
| toolOptions.addUniqueOptions(getInputFormatOptions()); |
| toolOptions.addUniqueOptions(getHiveOptions(true)); |
| toolOptions.addUniqueOptions(getHBaseOptions()); |
| toolOptions.addUniqueOptions(getHCatalogOptions()); |
| toolOptions.addUniqueOptions(getHCatImportOnlyOptions()); |
| toolOptions.addUniqueOptions(getAccumuloOptions()); |
| |
| // get common codegen opts. |
| RelatedOptions codeGenOpts = getCodeGenOpts(allTables); |
| |
| // add import-specific codegen opts: |
| codeGenOpts.addOption(OptionBuilder.withArgName("file") |
| .hasArg() |
| .withDescription("Disable code generation; use specified jar") |
| .withLongOpt(JAR_FILE_NAME_ARG) |
| .create()); |
| |
| toolOptions.addUniqueOptions(codeGenOpts); |
| } |
| |
| @Override |
| /** {@inheritDoc} */ |
| public void printHelp(ToolOptions toolOptions) { |
| super.printHelp(toolOptions); |
| System.out.println(""); |
| if (allTables) { |
| System.out.println("At minimum, you must specify --connect"); |
| } else { |
| System.out.println( |
| "At minimum, you must specify --connect and --table"); |
| } |
| |
| System.out.println( |
| "Arguments to mysqldump and other subprograms may be supplied"); |
| System.out.println( |
| "after a '--' on the command line."); |
| } |
| |
| private void applyIncrementalOptions(CommandLine in, SqoopOptions out) |
| throws InvalidOptionsException { |
| if (in.hasOption(INCREMENT_TYPE_ARG)) { |
| String incrementalTypeStr = in.getOptionValue(INCREMENT_TYPE_ARG); |
| if ("append".equals(incrementalTypeStr)) { |
| out.setIncrementalMode(SqoopOptions.IncrementalMode.AppendRows); |
| // This argument implies ability to append to the same directory. |
| out.setAppendMode(true); |
| } else if ("lastmodified".equals(incrementalTypeStr)) { |
| out.setIncrementalMode(SqoopOptions.IncrementalMode.DateLastModified); |
| } else { |
| throw new InvalidOptionsException("Unknown incremental import mode: " |
| + incrementalTypeStr + ". Use 'append' or 'lastmodified'." |
| + HELP_STR); |
| } |
| } |
| |
| if (in.hasOption(INCREMENT_COL_ARG)) { |
| out.setIncrementalTestColumn(in.getOptionValue(INCREMENT_COL_ARG)); |
| } |
| |
| if (in.hasOption(INCREMENT_LAST_VAL_ARG)) { |
| out.setIncrementalLastValue(in.getOptionValue(INCREMENT_LAST_VAL_ARG)); |
| } |
| } |
| |
| @Override |
| /** {@inheritDoc} */ |
| public void applyOptions(CommandLine in, SqoopOptions out) |
| throws InvalidOptionsException { |
| |
| try { |
| applyCommonOptions(in, out); |
| |
| if (in.hasOption(DIRECT_ARG)) { |
| out.setDirectMode(true); |
| } |
| |
| if (!allTables) { |
| if (in.hasOption(TABLE_ARG)) { |
| out.setTableName(in.getOptionValue(TABLE_ARG)); |
| } |
| |
| if (in.hasOption(COLUMNS_ARG)) { |
| String[] cols= in.getOptionValue(COLUMNS_ARG).split(","); |
| for (int i=0; i<cols.length; i++) { |
| cols[i] = cols[i].trim(); |
| } |
| out.setColumns(cols); |
| } |
| |
| if (in.hasOption(SPLIT_BY_ARG)) { |
| out.setSplitByCol(in.getOptionValue(SPLIT_BY_ARG)); |
| } |
| |
| if (in.hasOption(SPLIT_LIMIT_ARG)) { |
| out.setSplitLimit(Integer.parseInt(in.getOptionValue(SPLIT_LIMIT_ARG))); |
| } |
| |
| if (in.hasOption(WHERE_ARG)) { |
| out.setWhereClause(in.getOptionValue(WHERE_ARG)); |
| } |
| |
| if (in.hasOption(TARGET_DIR_ARG)) { |
| out.setTargetDir(in.getOptionValue(TARGET_DIR_ARG)); |
| } |
| |
| if (in.hasOption(APPEND_ARG)) { |
| out.setAppendMode(true); |
| } |
| |
| if (in.hasOption(DELETE_ARG)) { |
| out.setDeleteMode(true); |
| } |
| |
| if (in.hasOption(SQL_QUERY_ARG)) { |
| out.setSqlQuery(in.getOptionValue(SQL_QUERY_ARG)); |
| } |
| |
| if (in.hasOption(SQL_QUERY_BOUNDARY)) { |
| out.setBoundaryQuery(in.getOptionValue(SQL_QUERY_BOUNDARY)); |
| } |
| |
| if (in.hasOption(MERGE_KEY_ARG)) { |
| out.setMergeKeyCol(in.getOptionValue(MERGE_KEY_ARG)); |
| } |
| |
| applyValidationOptions(in, out); |
| } |
| |
| if (in.hasOption(WAREHOUSE_DIR_ARG)) { |
| out.setWarehouseDir(in.getOptionValue(WAREHOUSE_DIR_ARG)); |
| } |
| |
| if (in.hasOption(FMT_SEQUENCEFILE_ARG)) { |
| out.setFileLayout(SqoopOptions.FileLayout.SequenceFile); |
| } |
| |
| if (in.hasOption(FMT_TEXTFILE_ARG)) { |
| out.setFileLayout(SqoopOptions.FileLayout.TextFile); |
| } |
| |
| if (in.hasOption(FMT_AVRODATAFILE_ARG)) { |
| out.setFileLayout(SqoopOptions.FileLayout.AvroDataFile); |
| } |
| |
| if (in.hasOption(FMT_PARQUETFILE_ARG)) { |
| out.setFileLayout(SqoopOptions.FileLayout.ParquetFile); |
| } |
| |
| if (in.hasOption(NUM_MAPPERS_ARG)) { |
| out.setNumMappers(Integer.parseInt(in.getOptionValue(NUM_MAPPERS_ARG))); |
| } |
| |
| if (in.hasOption(MAPREDUCE_JOB_NAME)) { |
| out.setMapreduceJobName(in.getOptionValue(MAPREDUCE_JOB_NAME)); |
| } |
| |
| if (in.hasOption(COMPRESS_ARG)) { |
| out.setUseCompression(true); |
| } |
| |
| if (in.hasOption(COMPRESSION_CODEC_ARG)) { |
| out.setCompressionCodec(in.getOptionValue(COMPRESSION_CODEC_ARG)); |
| } |
| |
| if (in.hasOption(DIRECT_SPLIT_SIZE_ARG)) { |
| out.setDirectSplitSize(Long.parseLong(in.getOptionValue( |
| DIRECT_SPLIT_SIZE_ARG))); |
| } |
| |
| if (in.hasOption(INLINE_LOB_LIMIT_ARG)) { |
| out.setInlineLobLimit(Long.parseLong(in.getOptionValue( |
| INLINE_LOB_LIMIT_ARG))); |
| } |
| |
| if (in.hasOption(FETCH_SIZE_ARG)) { |
| out.setFetchSize(new Integer(in.getOptionValue(FETCH_SIZE_ARG))); |
| } |
| |
| if (in.hasOption(JAR_FILE_NAME_ARG)) { |
| out.setExistingJarName(in.getOptionValue(JAR_FILE_NAME_ARG)); |
| } |
| |
| if (in.hasOption(AUTORESET_TO_ONE_MAPPER)) { |
| out.setAutoResetToOneMapper(true); |
| } |
| |
| if (in.hasOption(ESCAPE_MAPPING_COLUMN_NAMES_ENABLED)) { |
| out.setEscapeMappingColumnNamesEnabled(Boolean.parseBoolean(in.getOptionValue( |
| ESCAPE_MAPPING_COLUMN_NAMES_ENABLED))); |
| } |
| |
| applyIncrementalOptions(in, out); |
| applyHiveOptions(in, out); |
| applyOutputFormatOptions(in, out); |
| applyInputFormatOptions(in, out); |
| applyCodeGenOptions(in, out, allTables); |
| applyHBaseOptions(in, out); |
| applyHCatalogOptions(in, out); |
| applyAccumuloOptions(in, out); |
| |
| } catch (NumberFormatException nfe) { |
| throw new InvalidOptionsException("Error: expected numeric argument.\n" |
| + "Try --help for usage."); |
| } |
| } |
| |
| /** |
| * Validate import-specific arguments. |
| * @param options the configured SqoopOptions to check |
| */ |
| protected void validateImportOptions(SqoopOptions options) |
| throws InvalidOptionsException { |
| if (!allTables && options.getTableName() == null |
| && options.getSqlQuery() == null) { |
| throw new InvalidOptionsException( |
| "--table or --" + SQL_QUERY_ARG + " is required for import. " |
| + "(Or use sqoop import-all-tables.)" |
| + HELP_STR); |
| } else if (options.getExistingJarName() != null |
| && options.getClassName() == null) { |
| throw new InvalidOptionsException("Jar specified with --jar-file, but no " |
| + "class specified with --class-name." + HELP_STR); |
| } else if (options.getTargetDir() != null |
| && options.getWarehouseDir() != null) { |
| throw new InvalidOptionsException( |
| "--target-dir with --warehouse-dir are incompatible options." |
| + HELP_STR); |
| } else if (options.getTableName() != null |
| && options.getSqlQuery() != null) { |
| throw new InvalidOptionsException( |
| "Cannot specify --" + SQL_QUERY_ARG + " and --table together." |
| + HELP_STR); |
| } else if (options.getSqlQuery() != null |
| && options.getTargetDir() == null |
| && options.getHBaseTable() == null |
| && options.getHCatTableName() == null |
| && options.getAccumuloTable() == null) { |
| throw new InvalidOptionsException( |
| "Must specify destination with --target-dir. " |
| + HELP_STR); |
| } else if (options.getSqlQuery() != null && options.doHiveImport() |
| && options.getHiveTableName() == null) { |
| throw new InvalidOptionsException( |
| "When importing a query to Hive, you must specify --" |
| + HIVE_TABLE_ARG + "." + HELP_STR); |
| } else if (options.getSqlQuery() != null && options.getNumMappers() > 1 |
| && options.getSplitByCol() == null) { |
| throw new InvalidOptionsException( |
| "When importing query results in parallel, you must specify --" |
| + SPLIT_BY_ARG + "." + HELP_STR); |
| } else if (options.isDirect()) { |
| validateDirectImportOptions(options); |
| } else if (allTables && options.isValidationEnabled()) { |
| throw new InvalidOptionsException("Validation is not supported for " |
| + "all tables but single table only."); |
| } else if (options.getSqlQuery() != null && options.isValidationEnabled()) { |
| throw new InvalidOptionsException("Validation is not supported for " |
| + "free from query but single table only."); |
| } else if (options.getWhereClause() != null |
| && options.isValidationEnabled()) { |
| throw new InvalidOptionsException("Validation is not supported for " |
| + "where clause but single table only."); |
| } else if (options.getIncrementalMode() |
| != SqoopOptions.IncrementalMode.None && options.isValidationEnabled()) { |
| throw new InvalidOptionsException("Validation is not supported for " |
| + "incremental imports but single table only."); |
| } else if ((options.getTargetDir() != null |
| || options.getWarehouseDir() != null) |
| && options.getHCatTableName() != null) { |
| throw new InvalidOptionsException("--hcatalog-table cannot be used " |
| + " --warehouse-dir or --target-dir options"); |
| } else if (options.isDeleteMode() && options.isAppendMode()) { |
| throw new InvalidOptionsException("--append and --delete-target-dir can" |
| + " not be used together."); |
| } else if (options.isDeleteMode() && options.getIncrementalMode() |
| != SqoopOptions.IncrementalMode.None) { |
| throw new InvalidOptionsException("--delete-target-dir can not be used" |
| + " with incremental imports."); |
| } else if (options.getAutoResetToOneMapper() |
| && (options.getSplitByCol() != null)) { |
| throw new InvalidOptionsException("--autoreset-to-one-mapper and" |
| + " --split-by cannot be used together."); |
| } |
| } |
| |
| void validateDirectImportOptions(SqoopOptions options) throws InvalidOptionsException { |
| validateDirectMysqlOptions(options); |
| validateDirectDropHiveDelimOption(options); |
| validateHasDirectConnectorOption(options); |
| } |
| |
| void validateDirectDropHiveDelimOption(SqoopOptions options) throws InvalidOptionsException { |
| if (options.doHiveDropDelims()) { |
| throw new InvalidOptionsException( |
| "Direct import currently do not support dropping hive delimiters," |
| + " please remove parameter --hive-drop-import-delims."); |
| } |
| } |
| |
| void validateDirectMysqlOptions(SqoopOptions options) throws InvalidOptionsException { |
| if (!MYSQL.isTheManagerTypeOf(options)) { |
| return; |
| } |
| if (options.getFileLayout() != SqoopOptions.FileLayout.TextFile) { |
| throw new InvalidOptionsException( |
| "MySQL direct import currently supports only text output format. " |
| + "Parameters --as-sequencefile --as-avrodatafile and --as-parquetfile are not " |
| + "supported with --direct params in MySQL case."); |
| } |
| if (options.getNullStringValue() != null || options.getNullNonStringValue() != null) { |
| throw new InvalidOptionsException( |
| "The --direct option is not compatible with the --null-string or " + |
| "--null-non-string command for MySQL imports"); |
| } |
| } |
| /** |
| * Validate the incremental import options. |
| */ |
| private void validateIncrementalOptions(SqoopOptions options) |
| throws InvalidOptionsException { |
| if (options.getIncrementalMode() != SqoopOptions.IncrementalMode.None |
| && options.getIncrementalTestColumn() == null) { |
| throw new InvalidOptionsException( |
| "For an incremental import, the check column must be specified " |
| + "with --" + INCREMENT_COL_ARG + ". " + HELP_STR); |
| } |
| |
| if (options.getIncrementalMode() == SqoopOptions.IncrementalMode.None |
| && options.getIncrementalTestColumn() != null) { |
| throw new InvalidOptionsException( |
| "You must specify an incremental import mode with --" |
| + INCREMENT_TYPE_ARG + ". " + HELP_STR); |
| } |
| |
| if (options.getIncrementalMode() == SqoopOptions.IncrementalMode.DateLastModified |
| && options.getFileLayout() == SqoopOptions.FileLayout.AvroDataFile) { |
| throw new InvalidOptionsException("--" |
| + INCREMENT_TYPE_ARG + " lastmodified cannot be used in conjunction with --" |
| + FMT_AVRODATAFILE_ARG + "." + HELP_STR); |
| } |
| } |
| |
| @Override |
| /** {@inheritDoc} */ |
| public void validateOptions(SqoopOptions options) |
| throws InvalidOptionsException { |
| |
| // If extraArguments is full, check for '--' followed by args for |
| // mysqldump or other commands we rely on. |
| options.setExtraArgs(getSubcommandArgs(extraArguments)); |
| int dashPos = getDashPosition(extraArguments); |
| |
| if (hasUnrecognizedArgs(extraArguments, 0, dashPos)) { |
| throw new InvalidOptionsException(HELP_STR); |
| } |
| |
| validateImportOptions(options); |
| validateIncrementalOptions(options); |
| validateCommonOptions(options); |
| validateCodeGenOptions(options); |
| validateOutputFormatOptions(options); |
| validateHBaseOptions(options); |
| validateHiveOptions(options); |
| validateHCatalogOptions(options); |
| validateAccumuloOptions(options); |
| } |
| } |
| |
