diff --git a/src/java/org/apache/sqoop/avro/AvroSchemaMismatchException.java b/src/java/org/apache/sqoop/avro/AvroSchemaMismatchException.java
new file mode 100644
index 0000000..4070627
--- /dev/null
+++ b/src/java/org/apache/sqoop/avro/AvroSchemaMismatchException.java
@@ -0,0 +1,50 @@
+/**
+ * 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
+ * <p>
+ * http://www.apache.org/licenses/LICENSE-2.0
+ * <p>
+ * 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.avro;
+
+import org.apache.avro.Schema;
+
+/**
+ * This exception will be thrown when Sqoop tries to write to a dataset
+ * and the Avro schema which was used when the dataset was created does not match
+ * the actual schema which is used by Sqoop during the write operation.
+ */
+public class AvroSchemaMismatchException extends RuntimeException {
+
+  static final String MESSAGE_TEMPLATE = "%s%nExpected schema: %s%nActual schema: %s";
+
+  private final Schema writtenWithSchema;
+
+  private final Schema actualSchema;
+
+  public AvroSchemaMismatchException(String message, Schema writtenWithSchema, Schema actualSchema) {
+    super(String.format(MESSAGE_TEMPLATE, message, writtenWithSchema.toString(), actualSchema.toString()));
+    this.writtenWithSchema = writtenWithSchema;
+    this.actualSchema = actualSchema;
+  }
+
+  public Schema getWrittenWithSchema() {
+    return writtenWithSchema;
+  }
+
+  public Schema getActualSchema() {
+    return actualSchema;
+  }
+
+}
diff --git a/src/java/org/apache/sqoop/mapreduce/ParquetJob.java b/src/java/org/apache/sqoop/mapreduce/ParquetJob.java
index b077d9b..4604773 100644
--- a/src/java/org/apache/sqoop/mapreduce/ParquetJob.java
+++ b/src/java/org/apache/sqoop/mapreduce/ParquetJob.java
@@ -28,6 +28,7 @@
 import org.apache.hadoop.mapreduce.security.token.delegation.DelegationTokenIdentifier;
 import org.apache.hadoop.security.UserGroupInformation;
 import org.apache.hadoop.security.token.Token;
+import org.apache.sqoop.avro.AvroSchemaMismatchException;
 import org.apache.sqoop.hive.HiveConfig;
 import org.kitesdk.data.CompressionType;
 import org.kitesdk.data.Dataset;
@@ -48,12 +49,23 @@
   public static final Log LOG = LogFactory.getLog(ParquetJob.class.getName());
 
   public static final String HIVE_METASTORE_CLIENT_CLASS = "org.apache.hadoop.hive.metastore.HiveMetaStoreClient";
+
   public static final String HIVE_METASTORE_SASL_ENABLED = "hive.metastore.sasl.enabled";
   // Purposefully choosing the same token alias as the one Oozie chooses.
   // Make sure we don't generate a new delegation token if oozie
   // has already generated one.
   public static final String HIVE_METASTORE_TOKEN_ALIAS = "HCat Token";
 
+  public static final String INCOMPATIBLE_AVRO_SCHEMA_MSG = "Target dataset was created with an incompatible Avro schema. ";
+
+  public static final String HIVE_INCOMPATIBLE_AVRO_SCHEMA_MSG = "You tried to import to an already existing Hive table in " +
+      "Parquet format. Sqoop maps date/timestamp SQL types to int/bigint Hive types during Hive Parquet import" +
+      " but it is possible that date/timestamp types were mapped to strings during table" +
+      " creation. Consider using Sqoop option --map-column-java resolve the mismatch" +
+      " (e.g. --map-column-java date_field1=String,timestamp_field1=String).";
+
+  private static final String HIVE_URI_PREFIX = "dataset:hive";
+
   private ParquetJob() {
   }
 
@@ -91,7 +103,7 @@
     Dataset dataset;
 
     // Add hive delegation token only if we don't already have one.
-    if (uri.startsWith("dataset:hive")) {
+    if (isHiveImport(uri)) {
       Configuration hiveConf = HiveConfig.getHiveConf(conf);
       if (isSecureMetastore(hiveConf)) {
         // Copy hive configs to job config
@@ -111,9 +123,8 @@
       dataset = Datasets.load(uri);
       Schema writtenWith = dataset.getDescriptor().getSchema();
       if (!SchemaValidationUtil.canRead(writtenWith, schema)) {
-        throw new IOException(
-            String.format("Expected schema: %s%nActual schema: %s",
-                writtenWith, schema));
+        String exceptionMessage = buildAvroSchemaMismatchMessage(isHiveImport(uri));
+        throw new AvroSchemaMismatchException(exceptionMessage, writtenWith, schema);
       }
     } else {
       dataset = createDataset(schema, getCompressionType(conf), uri);
@@ -131,7 +142,11 @@
     }
   }
 
-  private static Dataset createDataset(Schema schema,
+  private static boolean isHiveImport(String importUri) {
+    return importUri.startsWith(HIVE_URI_PREFIX);
+  }
+
+  public static Dataset createDataset(Schema schema,
       CompressionType compressionType, String uri) {
     DatasetDescriptor descriptor = new DatasetDescriptor.Builder()
         .schema(schema)
@@ -191,4 +206,15 @@
       throw new RuntimeException("Couldn't fetch delegation token.", ex);
     }
   }
+
+  private static String buildAvroSchemaMismatchMessage(boolean hiveImport) {
+    String exceptionMessage = INCOMPATIBLE_AVRO_SCHEMA_MSG;
+
+    if (hiveImport) {
+      exceptionMessage += HIVE_INCOMPATIBLE_AVRO_SCHEMA_MSG;
+    }
+
+    return exceptionMessage;
+  }
+
 }
diff --git a/src/java/org/apache/sqoop/tool/ImportTool.java b/src/java/org/apache/sqoop/tool/ImportTool.java
index ed951ea..258ef79 100644
--- a/src/java/org/apache/sqoop/tool/ImportTool.java
+++ b/src/java/org/apache/sqoop/tool/ImportTool.java
@@ -52,7 +52,7 @@
 import com.cloudera.sqoop.metastore.JobStorageFactory;
 import com.cloudera.sqoop.util.AppendUtils;
 import com.cloudera.sqoop.util.ImportException;
-import org.apache.sqoop.manager.SupportedManagers;
+import org.apache.sqoop.avro.AvroSchemaMismatchException;
 
 import static org.apache.sqoop.manager.SupportedManagers.MYSQL;
 
@@ -63,6 +63,8 @@
 
   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
@@ -81,8 +83,12 @@
   }
 
   public ImportTool(String toolName, boolean allTables) {
+    this(toolName, new CodeGenTool(), allTables);
+  }
+
+  public ImportTool(String toolName, CodeGenTool codeGenerator, boolean allTables) {
     super(toolName);
-    this.codeGenerator = new CodeGenTool();
+    this.codeGenerator = codeGenerator;
     this.allTables = allTables;
   }
 
@@ -616,18 +622,21 @@
       // Import a single table (or query) the user specified.
       importTable(options, options.getTableName(), hiveImport);
     } catch (IllegalArgumentException iea) {
-        LOG.error("Imported Failed: " + iea.getMessage());
+        LOG.error(IMPORT_FAILED_ERROR_MSG + iea.getMessage());
       rethrowIfRequired(options, iea);
       return 1;
     } catch (IOException ioe) {
-      LOG.error("Encountered IOException running import job: "
-          + StringUtils.stringifyException(ioe));
+      LOG.error(IMPORT_FAILED_ERROR_MSG + StringUtils.stringifyException(ioe));
       rethrowIfRequired(options, ioe);
       return 1;
     } catch (ImportException ie) {
-      LOG.error("Error during import: " + ie.toString());
+      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);
     }
diff --git a/src/test/com/cloudera/sqoop/hive/TestHiveImport.java b/src/test/com/cloudera/sqoop/hive/TestHiveImport.java
index 6f488ab..1253e8d 100644
--- a/src/test/com/cloudera/sqoop/hive/TestHiveImport.java
+++ b/src/test/com/cloudera/sqoop/hive/TestHiveImport.java
@@ -26,13 +26,18 @@
 import java.util.Arrays;
 import java.util.List;
 
+import com.cloudera.sqoop.Sqoop;
 import junit.framework.JUnit4TestAdapter;
+import org.apache.avro.Schema;
+import org.apache.avro.SchemaBuilder;
 import org.apache.avro.generic.GenericRecord;
 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.avro.AvroSchemaMismatchException;
+import org.apache.sqoop.mapreduce.ParquetJob;
 import org.junit.After;
 import org.junit.Before;
 import org.junit.Rule;
@@ -54,6 +59,8 @@
 import org.kitesdk.data.Dataset;
 import org.kitesdk.data.DatasetReader;
 import org.kitesdk.data.Datasets;
+import org.kitesdk.data.Formats;
+import org.kitesdk.data.spi.DefaultConfiguration;
 
 /**
  * Test HiveImport capability after an import to HDFS.
@@ -388,6 +395,42 @@
     verifyHiveDataset(TABLE_NAME, new Object[][] {{"test2", 24, "somestring2"}});
   }
 
+  @Test
+  public void testHiveImportAsParquetWhenTableExistsWithIncompatibleSchema() throws Exception {
+    final String TABLE_NAME = "HIVE_IMPORT_AS_PARQUET_EXISTING_TABLE";
+    setCurTableName(TABLE_NAME);
+    setNumCols(3);
+
+    String [] types = { "VARCHAR(32)", "INTEGER", "DATE" };
+    String [] vals = { "'test'", "42", "'2009-12-31'" };
+    String [] extraArgs = {"--as-parquetfile"};
+
+    createHiveDataSet(TABLE_NAME);
+
+    createTableWithColTypes(types, vals);
+
+    thrown.expect(AvroSchemaMismatchException.class);
+    thrown.expectMessage(ParquetJob.INCOMPATIBLE_AVRO_SCHEMA_MSG + ParquetJob.HIVE_INCOMPATIBLE_AVRO_SCHEMA_MSG);
+
+    SqoopOptions sqoopOptions = getSqoopOptions(getConf());
+    sqoopOptions.setThrowOnError(true);
+    Sqoop sqoop = new Sqoop(new ImportTool(), getConf(), sqoopOptions);
+    sqoop.run(getArgv(false, extraArgs));
+
+  }
+
+  private void createHiveDataSet(String tableName) {
+    Schema dataSetSchema = SchemaBuilder
+        .record(tableName)
+            .fields()
+            .name(getColName(0)).type().nullable().stringType().noDefault()
+            .name(getColName(1)).type().nullable().stringType().noDefault()
+            .name(getColName(2)).type().nullable().stringType().noDefault()
+            .endRecord();
+    String dataSetUri = "dataset:hive:/default/" + tableName;
+    ParquetJob.createDataset(dataSetSchema, Formats.PARQUET.getDefaultCompressionType(), dataSetUri);
+  }
+
   /**
    * Test that records are appended to an existing table.
    */
diff --git a/src/test/org/apache/sqoop/tool/TestImportTool.java b/src/test/org/apache/sqoop/tool/TestImportTool.java
index 4136e9f..7e11f54 100644
--- a/src/test/org/apache/sqoop/tool/TestImportTool.java
+++ b/src/test/org/apache/sqoop/tool/TestImportTool.java
@@ -20,11 +20,25 @@
 
 import static org.hamcrest.CoreMatchers.equalTo;
 import static org.hamcrest.CoreMatchers.is;
+import static org.junit.Assert.assertEquals;
 import static org.junit.Assert.assertThat;
+import static org.mockito.Matchers.any;
+import static org.mockito.Matchers.anyString;
+import static org.mockito.Mockito.doReturn;
+import static org.mockito.Mockito.doThrow;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.spy;
+import static org.mockito.Mockito.when;
 
 import java.sql.Connection;
 
+import com.cloudera.sqoop.hive.HiveImport;
+import org.apache.avro.Schema;
 import org.apache.sqoop.SqoopOptions;
+import org.apache.sqoop.avro.AvroSchemaMismatchException;
+import org.apache.sqoop.util.ExpectedLogMessage;
+import org.junit.Rule;
+import org.junit.Test;
 import org.junit.experimental.theories.DataPoints;
 import org.junit.experimental.theories.Theories;
 import org.junit.experimental.theories.Theory;
@@ -41,6 +55,9 @@
     {"TRANSACTION_SERIALIZABLE",Connection.TRANSACTION_SERIALIZABLE}
   };
 
+  @Rule
+  public ExpectedLogMessage logMessage = new ExpectedLogMessage();
+
   @Theory
   public void esnureTransactionIsolationLevelsAreMappedToTheRightValues(Object[] values)
       throws Exception {
@@ -50,4 +67,30 @@
     assertThat(options.getMetadataTransactionIsolationLevel(), is(equalTo(values[1])));
   }
 
+  @Test
+  public void testImportToolHandlesAvroSchemaMismatchExceptionProperly() throws Exception {
+    final String writtenWithSchemaString = "writtenWithSchema";
+    final String actualSchemaString = "actualSchema";
+    final String errorMessage = "Import failed";
+
+    ImportTool importTool = spy(new ImportTool("import", mock(CodeGenTool.class), false));
+
+    doReturn(true).when(importTool).init(any(com.cloudera.sqoop.SqoopOptions.class));
+
+    Schema writtenWithSchema = mock(Schema.class);
+    when(writtenWithSchema.toString()).thenReturn(writtenWithSchemaString);
+    Schema actualSchema = mock(Schema.class);
+    when(actualSchema.toString()).thenReturn(actualSchemaString);
+
+    AvroSchemaMismatchException expectedException = new AvroSchemaMismatchException(errorMessage, writtenWithSchema, actualSchema);
+    doThrow(expectedException).when(importTool).importTable(any(com.cloudera.sqoop.SqoopOptions.class), anyString(), any(HiveImport.class));
+
+    com.cloudera.sqoop.SqoopOptions sqoopOptions = mock(com.cloudera.sqoop.SqoopOptions.class);
+    when(sqoopOptions.doHiveImport()).thenReturn(true);
+
+    logMessage.expectError(expectedException.getMessage());
+    int result = importTool.run(sqoopOptions);
+    assertEquals(1, result);
+  }
+
 }
diff --git a/src/test/org/apache/sqoop/util/ExpectedLogMessage.java b/src/test/org/apache/sqoop/util/ExpectedLogMessage.java
new file mode 100644
index 0000000..0372fe2
--- /dev/null
+++ b/src/test/org/apache/sqoop/util/ExpectedLogMessage.java
@@ -0,0 +1,147 @@
+/**
+ * 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
+ * <p>
+ * http://www.apache.org/licenses/LICENSE-2.0
+ * <p>
+ * 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.util;
+
+import org.apache.log4j.Appender;
+import org.apache.log4j.Level;
+import org.apache.log4j.Logger;
+import org.apache.log4j.spi.LoggingEvent;
+import org.apache.log4j.spi.ThrowableInformation;
+import org.hamcrest.Matcher;
+import org.hamcrest.TypeSafeMatcher;
+import org.junit.rules.TestRule;
+import org.junit.runner.Description;
+import org.junit.runners.model.Statement;
+import org.mockito.ArgumentCaptor;
+
+import static org.apache.commons.lang.StringUtils.EMPTY;
+import static org.apache.commons.lang.StringUtils.contains;
+import static org.apache.commons.lang.StringUtils.defaultString;
+import static org.hamcrest.CoreMatchers.hasItem;
+import static org.junit.Assert.assertThat;
+import static org.mockito.Mockito.atMost;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.verify;
+
+public class ExpectedLogMessage implements TestRule {
+
+  private static class LoggingEventMatcher extends TypeSafeMatcher<LoggingEvent> {
+
+    private final String msg;
+
+    private final Level level;
+
+    private LoggingEventMatcher(String msg, Level level) {
+      this.msg = msg;
+      this.level = level;
+    }
+
+    @Override
+    public boolean matchesSafely(LoggingEvent o) {
+      return contains(extractEventMessage(o), msg) && level.equals(o.getLevel());
+    }
+
+    @Override
+    public void describeTo(org.hamcrest.Description description) {
+      description.appendText(eventToString(msg, level));
+    }
+
+    @Override
+    protected void describeMismatchSafely(LoggingEvent item, org.hamcrest.Description mismatchDescription) {
+      mismatchDescription.appendText(eventToString(extractEventMessage(item), item.getLevel()));
+    }
+
+    private String extractEventMessage(LoggingEvent item) {
+      final String eventMsg = item.getRenderedMessage();
+      final String exceptionMessage = extractExceptionMessage(item.getThrowableInformation());
+
+      return eventMsg + exceptionMessage;
+    }
+
+    private String extractExceptionMessage(ThrowableInformation throwableInfo) {
+      if (throwableInfo == null) {
+        return EMPTY;
+      }
+
+      Throwable throwable = throwableInfo.getThrowable();
+      if (throwable == null) {
+        return EMPTY;
+      }
+
+      return defaultString(throwable.getMessage());
+    }
+
+    private String eventToString(String msg, Level level) {
+      return "Log entry [ " + msg + ", " + level + " ]";
+    }
+
+  }
+
+  private Matcher<LoggingEvent> loggingEventMatcher;
+
+  @Override
+  public Statement apply(final Statement base, Description description) {
+    return new Statement() {
+      @Override
+      public void evaluate() throws Throwable {
+
+        Logger rootLogger = Logger.getRootLogger();
+        Appender mockAppender = mock(Appender.class);
+        rootLogger.addAppender(mockAppender);
+
+        try {
+          base.evaluate();
+          if (loggingEventMatcher != null) {
+            ArgumentCaptor<LoggingEvent> argumentCaptor = ArgumentCaptor.forClass(LoggingEvent.class);
+            verify(mockAppender, atMost(Integer.MAX_VALUE)).doAppend(argumentCaptor.capture());
+            assertThat(argumentCaptor.getAllValues(), hasItem(loggingEventMatcher));
+          }
+        } finally {
+          rootLogger.removeAppender(mockAppender);
+          loggingEventMatcher = null;
+        }
+      }
+    };
+  }
+
+  public void expectFatal(String msg) {
+    loggingEventMatcher = new LoggingEventMatcher(msg, Level.FATAL);
+  }
+
+  public void expectError(String msg) {
+    loggingEventMatcher = new LoggingEventMatcher(msg, Level.ERROR);
+  }
+
+  public void expectWarn(String msg) {
+    loggingEventMatcher = new LoggingEventMatcher(msg, Level.WARN);
+  }
+
+  public void expectInfo(String msg) {
+    loggingEventMatcher = new LoggingEventMatcher(msg, Level.INFO);
+  }
+
+  public void expectDebug(String msg) {
+    loggingEventMatcher = new LoggingEventMatcher(msg, Level.DEBUG);
+  }
+
+  public void expectTrace(String msg) {
+    loggingEventMatcher = new LoggingEventMatcher(msg, Level.TRACE);
+  }
+
+}