src/java/org/apache/sqoop/hive/TableDefWriter.java - sqoop - Git at Google

 /**
  * 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.hive;

 import java.io.File;
 import java.io.IOException;
 import java.util.Map;
 import java.util.Date;
 import java.text.DateFormat;
 import java.text.SimpleDateFormat;
 import java.util.Properties;

 import org.apache.commons.lang.StringUtils;
 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.sqoop.io.CodecMap;

 import org.apache.sqoop.SqoopOptions;
 import org.apache.sqoop.manager.ConnManager;
 import org.apache.sqoop.util.FileSystemUtil;

 /**
  * Creates (Hive-specific) SQL DDL statements to create tables to hold data
  * we're importing from another source.
  *
  * After we import the database into HDFS, we can inject it into Hive using
  * the CREATE TABLE and LOAD DATA INPATH statements generated by this object.
  */
 public class TableDefWriter {

   public static final Log LOG = LogFactory.getLog(
       TableDefWriter.class.getName());

   private SqoopOptions options;
   private ConnManager connManager;
   private Configuration configuration;
   private String inputTableName;
   private String outputTableName;
   private boolean commentsEnabled;

   /**
    * Creates a new TableDefWriter to generate a Hive CREATE TABLE statement.
    * @param opts program-wide options
    * @param connMgr the connection manager used to describe the table.
    * @param inputTable the name of the table to load.
    * @param outputTable the name of the Hive table to create.
    * @param config the Hadoop configuration to use to connect to the dfs
    * @param withComments if true, then tables will be created with a
    *        timestamp comment.
    */
   public TableDefWriter(final SqoopOptions opts, final ConnManager connMgr,
       final String inputTable, final String outputTable,
       final Configuration config, final boolean withComments) {
     this.options = opts;
     this.connManager = connMgr;
     this.inputTableName = inputTable;
     this.outputTableName = outputTable;
     this.configuration = config;
     this.commentsEnabled = withComments;
   }

   /**
    * Get the column names to import.
    */
   private String [] getColumnNames() {
     String [] colNames = options.getColumns();
     if (null != colNames) {
       return colNames; // user-specified column names.
     } else if (null != inputTableName) {
       return connManager.getColumnNames(inputTableName);
     } else {
       return connManager.getColumnNamesForQuery(options.getSqlQuery());
     }
   }

   /**
    * @return the CREATE TABLE statement for the table to load into hive.
    */
   public String getCreateTableStmt() throws IOException {
     resetConnManager();
     Map<String, Integer> columnTypes;
     Properties userMapping = options.getMapColumnHive();
     Boolean isHiveExternalTableSet = !StringUtils.isBlank(options.getHiveExternalTableDir());
     // Get these from the database.
     if (null != inputTableName) {
       columnTypes = connManager.getColumnTypes(inputTableName);
     } else {
       columnTypes = connManager.getColumnTypesForQuery(options.getSqlQuery());
     }

     String [] colNames = getColumnNames();
     StringBuilder sb = new StringBuilder();
     if (options.doFailIfHiveTableExists()) {
       if (isHiveExternalTableSet) {
         sb.append("CREATE EXTERNAL TABLE `");
       } else {
         sb.append("CREATE TABLE `");
       }
     } else {
       if (isHiveExternalTableSet) {
         sb.append("CREATE EXTERNAL TABLE IF NOT EXISTS `");
       } else {
         sb.append("CREATE TABLE IF NOT EXISTS `");
       }
     }

     if(options.getHiveDatabaseName() != null) {
       sb.append(options.getHiveDatabaseName()).append("`.`");
     }
     sb.append(outputTableName).append("` ( ");

     // Check that all explicitly mapped columns are present in result set
     for(Object column : userMapping.keySet()) {
       boolean found = false;
       for(String c : colNames) {
         if (c.equals(column)) {
           found = true;
           break;
         }
       }

       if (!found) {
         throw new IllegalArgumentException("No column by the name " + column
                 + "found while importing data");
       }
     }

     boolean first = true;
     String partitionKey = options.getHivePartitionKey();
     for (String col : colNames) {
       if (col.equals(partitionKey)) {
         throw new IllegalArgumentException("Partition key " + col + " cannot "
             + "be a column to import.");
       }

       if (!first) {
         sb.append(", ");
       }

       first = false;

       Integer colType = columnTypes.get(col);
       String hiveColType = userMapping.getProperty(col);
       if (hiveColType == null) {
         hiveColType = connManager.toHiveType(inputTableName, col, colType);
       }
       if (null == hiveColType) {
         throw new IOException("Hive does not support the SQL type for column "
             + col);
       }

       sb.append('`').append(col).append("` ").append(hiveColType);

       if (HiveTypes.isHiveTypeImprovised(colType)) {
         LOG.warn(
             "Column " + col + " had to be cast to a less precise type in Hive");
       }
     }

     sb.append(") ");

     if (commentsEnabled) {
       DateFormat dateFormat = new SimpleDateFormat("yyyy/MM/dd HH:mm:ss");
       String curDateStr = dateFormat.format(new Date());
       sb.append("COMMENT 'Imported by sqoop on " + curDateStr + "' ");
     }

     if (partitionKey != null) {
       sb.append("PARTITIONED BY (")
         .append(partitionKey)
         .append(" STRING) ");
      }

     sb.append("ROW FORMAT DELIMITED FIELDS TERMINATED BY '");
     sb.append(getHiveOctalCharCode((int) options.getOutputFieldDelim()));
     sb.append("' LINES TERMINATED BY '");
     sb.append(getHiveOctalCharCode((int) options.getOutputRecordDelim()));
     String codec = options.getCompressionCodec();
     if (codec != null && (codec.equals(CodecMap.LZOP)
             || codec.equals(CodecMap.getCodecClassName(CodecMap.LZOP)))) {
       sb.append("' STORED AS INPUTFORMAT "
               + "'com.hadoop.mapred.DeprecatedLzoTextInputFormat'");
       sb.append(" OUTPUTFORMAT "
               + "'org.apache.hadoop.hive.ql.io.HiveIgnoreKeyTextOutputFormat'");
     } else {
       sb.append("' STORED AS TEXTFILE");
     }

     if (isHiveExternalTableSet) {
       // add location
       sb.append(" LOCATION '"+options.getHiveExternalTableDir()+"'");
     }

     LOG.debug("Create statement: " + sb.toString());
     return sb.toString();
   }

   /**
    * @return the LOAD DATA statement to import the data in HDFS into hive.
    */
   public String getLoadDataStmt() throws IOException {
     Path finalPath = getFinalPath();

     StringBuilder sb = new StringBuilder();
     sb.append("LOAD DATA INPATH '");
     sb.append(finalPath.toString() + "'");
     if (options.doOverwriteHiveTable()) {
       sb.append(" OVERWRITE");
     }
     sb.append(" INTO TABLE `");
     if(options.getHiveDatabaseName() != null) {
       sb.append(options.getHiveDatabaseName()).append("`.`");
     }
     sb.append(outputTableName);
     sb.append('`');

     if (options.getHivePartitionKey() != null) {
       sb.append(" PARTITION (")
         .append(options.getHivePartitionKey())
         .append("='").append(options.getHivePartitionValue())
         .append("')");
     }

     LOG.debug("Load statement: " + sb.toString());
     return sb.toString();
   }

   public Path getFinalPath() throws IOException {
     String warehouseDir = options.getWarehouseDir();
     if (null == warehouseDir) {
       warehouseDir = "";
     } else if (!warehouseDir.endsWith(File.separator)) {
       warehouseDir = warehouseDir + File.separator;
     }

     // Final path is determined in the following order:
     // 1. Use target dir if the user specified.
     // 2. Use input table name.
     String tablePath = null;
     String targetDir = options.getTargetDir();
     if (null != targetDir) {
       tablePath = warehouseDir + targetDir;
     } else {
       tablePath = warehouseDir + inputTableName;
     }
     return FileSystemUtil.makeQualified(new Path(tablePath), configuration);
   }

   /**
    * Return a string identifying the character to use as a delimiter
    * in Hive, in octal representation.
    * Hive can specify delimiter characters in the form '\ooo' where
    * ooo is a three-digit octal number between 000 and 177. Values
    * may not be truncated ('\12' is wrong; '\012' is ok) nor may they
    * be zero-prefixed (e.g., '\0177' is wrong).
    *
    * @param charNum the character to use as a delimiter
    * @return a string of the form "\ooo" where ooo is an octal number
    * in [000, 177].
    * @throws IllegalArgumentException if charNum &gt; 0177.
    */
   public static String getHiveOctalCharCode(int charNum) {
     if (charNum > 0177) {
       throw new IllegalArgumentException(
           "Character " + charNum + " is an out-of-range delimiter");
     }

     return String.format("\\%03o", charNum);
   }

   /**
    * The JDBC connection owned by the ConnManager has been most probably opened when the import was started
    * so it might have timed out by the time TableDefWriter methods are invoked which happens at the end of import.
    * The task of this method is to discard the current connection held by ConnManager to make sure
    * that TableDefWriter will have a working one.
    */
   private void resetConnManager() {
     this.connManager.discardConnection(true);
   }

   SqoopOptions getOptions() {
     return options;
   }

   ConnManager getConnManager() {
     return connManager;
   }

   Configuration getConfiguration() {
     return configuration;
   }

   String getInputTableName() {
     return inputTableName;
   }

   String getOutputTableName() {
     return outputTableName;
   }

   boolean isCommentsEnabled() {
     return commentsEnabled;
   }
 }
	/**
	* 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.hive;

	import java.io.File;
	import java.io.IOException;
	import java.util.Map;
	import java.util.Date;
	import java.text.DateFormat;
	import java.text.SimpleDateFormat;
	import java.util.Properties;

	import org.apache.commons.lang.StringUtils;
	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.sqoop.io.CodecMap;

	import org.apache.sqoop.SqoopOptions;
	import org.apache.sqoop.manager.ConnManager;
	import org.apache.sqoop.util.FileSystemUtil;

	/**
	* Creates (Hive-specific) SQL DDL statements to create tables to hold data
	* we're importing from another source.
	*
	* After we import the database into HDFS, we can inject it into Hive using
	* the CREATE TABLE and LOAD DATA INPATH statements generated by this object.
	*/
	public class TableDefWriter {

	public static final Log LOG = LogFactory.getLog(
	TableDefWriter.class.getName());

	private SqoopOptions options;
	private ConnManager connManager;
	private Configuration configuration;
	private String inputTableName;
	private String outputTableName;
	private boolean commentsEnabled;

	/**
	* Creates a new TableDefWriter to generate a Hive CREATE TABLE statement.
	* @param opts program-wide options
	* @param connMgr the connection manager used to describe the table.
	* @param inputTable the name of the table to load.
	* @param outputTable the name of the Hive table to create.
	* @param config the Hadoop configuration to use to connect to the dfs
	* @param withComments if true, then tables will be created with a
	* timestamp comment.
	*/
	public TableDefWriter(final SqoopOptions opts, final ConnManager connMgr,
	final String inputTable, final String outputTable,
	final Configuration config, final boolean withComments) {
	this.options = opts;
	this.connManager = connMgr;
	this.inputTableName = inputTable;
	this.outputTableName = outputTable;
	this.configuration = config;
	this.commentsEnabled = withComments;
	}

	/**
	* Get the column names to import.
	*/
	private String [] getColumnNames() {
	String [] colNames = options.getColumns();
	if (null != colNames) {
	return colNames; // user-specified column names.
	} else if (null != inputTableName) {
	return connManager.getColumnNames(inputTableName);
	} else {
	return connManager.getColumnNamesForQuery(options.getSqlQuery());
	}
	}

	/**
	* @return the CREATE TABLE statement for the table to load into hive.
	*/
	public String getCreateTableStmt() throws IOException {
	resetConnManager();
	Map<String, Integer> columnTypes;
	Properties userMapping = options.getMapColumnHive();
	Boolean isHiveExternalTableSet = !StringUtils.isBlank(options.getHiveExternalTableDir());
	// Get these from the database.
	if (null != inputTableName) {
	columnTypes = connManager.getColumnTypes(inputTableName);
	} else {
	columnTypes = connManager.getColumnTypesForQuery(options.getSqlQuery());
	}

	String [] colNames = getColumnNames();
	StringBuilder sb = new StringBuilder();
	if (options.doFailIfHiveTableExists()) {
	if (isHiveExternalTableSet) {
	sb.append("CREATE EXTERNAL TABLE `");
	} else {
	sb.append("CREATE TABLE `");
	}
	} else {
	if (isHiveExternalTableSet) {
	sb.append("CREATE EXTERNAL TABLE IF NOT EXISTS `");
	} else {
	sb.append("CREATE TABLE IF NOT EXISTS `");
	}
	}

	if(options.getHiveDatabaseName() != null) {
	sb.append(options.getHiveDatabaseName()).append("`.`");
	}
	sb.append(outputTableName).append("` ( ");

	// Check that all explicitly mapped columns are present in result set
	for(Object column : userMapping.keySet()) {
	boolean found = false;
	for(String c : colNames) {
	if (c.equals(column)) {
	found = true;
	break;
	}
	}

	if (!found) {
	throw new IllegalArgumentException("No column by the name " + column
	+ "found while importing data");
	}
	}

	boolean first = true;
	String partitionKey = options.getHivePartitionKey();
	for (String col : colNames) {
	if (col.equals(partitionKey)) {
	throw new IllegalArgumentException("Partition key " + col + " cannot "
	+ "be a column to import.");
	}

	if (!first) {
	sb.append(", ");
	}

	first = false;

	Integer colType = columnTypes.get(col);
	String hiveColType = userMapping.getProperty(col);
	if (hiveColType == null) {
	hiveColType = connManager.toHiveType(inputTableName, col, colType);
	}
	if (null == hiveColType) {
	throw new IOException("Hive does not support the SQL type for column "
	+ col);
	}

	sb.append('`').append(col).append("` ").append(hiveColType);

	if (HiveTypes.isHiveTypeImprovised(colType)) {
	LOG.warn(
	"Column " + col + " had to be cast to a less precise type in Hive");
	}
	}

	sb.append(") ");

	if (commentsEnabled) {
	DateFormat dateFormat = new SimpleDateFormat("yyyy/MM/dd HH:mm:ss");
	String curDateStr = dateFormat.format(new Date());
	sb.append("COMMENT 'Imported by sqoop on " + curDateStr + "' ");
	}

	if (partitionKey != null) {
	sb.append("PARTITIONED BY (")
	.append(partitionKey)
	.append(" STRING) ");
	}

	sb.append("ROW FORMAT DELIMITED FIELDS TERMINATED BY '");
	sb.append(getHiveOctalCharCode((int) options.getOutputFieldDelim()));
	sb.append("' LINES TERMINATED BY '");
	sb.append(getHiveOctalCharCode((int) options.getOutputRecordDelim()));
	String codec = options.getCompressionCodec();
	if (codec != null && (codec.equals(CodecMap.LZOP)
	\|\| codec.equals(CodecMap.getCodecClassName(CodecMap.LZOP)))) {
	sb.append("' STORED AS INPUTFORMAT "
	+ "'com.hadoop.mapred.DeprecatedLzoTextInputFormat'");
	sb.append(" OUTPUTFORMAT "
	+ "'org.apache.hadoop.hive.ql.io.HiveIgnoreKeyTextOutputFormat'");
	} else {
	sb.append("' STORED AS TEXTFILE");
	}

	if (isHiveExternalTableSet) {
	// add location
	sb.append(" LOCATION '"+options.getHiveExternalTableDir()+"'");
	}

	LOG.debug("Create statement: " + sb.toString());
	return sb.toString();
	}

	/**
	* @return the LOAD DATA statement to import the data in HDFS into hive.
	*/
	public String getLoadDataStmt() throws IOException {
	Path finalPath = getFinalPath();

	StringBuilder sb = new StringBuilder();
	sb.append("LOAD DATA INPATH '");
	sb.append(finalPath.toString() + "'");
	if (options.doOverwriteHiveTable()) {
	sb.append(" OVERWRITE");
	}
	sb.append(" INTO TABLE `");
	if(options.getHiveDatabaseName() != null) {
	sb.append(options.getHiveDatabaseName()).append("`.`");
	}
	sb.append(outputTableName);
	sb.append('`');

	if (options.getHivePartitionKey() != null) {
	sb.append(" PARTITION (")
	.append(options.getHivePartitionKey())
	.append("='").append(options.getHivePartitionValue())
	.append("')");
	}

	LOG.debug("Load statement: " + sb.toString());
	return sb.toString();
	}

	public Path getFinalPath() throws IOException {
	String warehouseDir = options.getWarehouseDir();
	if (null == warehouseDir) {
	warehouseDir = "";
	} else if (!warehouseDir.endsWith(File.separator)) {
	warehouseDir = warehouseDir + File.separator;
	}

	// Final path is determined in the following order:
	// 1. Use target dir if the user specified.
	// 2. Use input table name.
	String tablePath = null;
	String targetDir = options.getTargetDir();
	if (null != targetDir) {
	tablePath = warehouseDir + targetDir;
	} else {
	tablePath = warehouseDir + inputTableName;
	}
	return FileSystemUtil.makeQualified(new Path(tablePath), configuration);
	}

	/**
	* Return a string identifying the character to use as a delimiter
	* in Hive, in octal representation.
	* Hive can specify delimiter characters in the form '\ooo' where
	* ooo is a three-digit octal number between 000 and 177. Values
	* may not be truncated ('\12' is wrong; '\012' is ok) nor may they
	* be zero-prefixed (e.g., '\0177' is wrong).
	*
	* @param charNum the character to use as a delimiter
	* @return a string of the form "\ooo" where ooo is an octal number
	* in [000, 177].
	* @throws IllegalArgumentException if charNum > 0177.
	*/
	public static String getHiveOctalCharCode(int charNum) {
	if (charNum > 0177) {
	throw new IllegalArgumentException(
	"Character " + charNum + " is an out-of-range delimiter");
	}

	return String.format("\\%03o", charNum);
	}

	/**
	* The JDBC connection owned by the ConnManager has been most probably opened when the import was started
	* so it might have timed out by the time TableDefWriter methods are invoked which happens at the end of import.
	* The task of this method is to discard the current connection held by ConnManager to make sure
	* that TableDefWriter will have a working one.
	*/
	private void resetConnManager() {
	this.connManager.discardConnection(true);
	}

	SqoopOptions getOptions() {
	return options;
	}

	ConnManager getConnManager() {
	return connManager;
	}

	Configuration getConfiguration() {
	return configuration;
	}

	String getInputTableName() {
	return inputTableName;
	}

	String getOutputTableName() {
	return outputTableName;
	}

	boolean isCommentsEnabled() {
	return commentsEnabled;
	}
	}