diff --git a/src/docs/man/hbase-args.txt b/src/docs/man/hbase-args.txt
index afd5c5b..af9c96a 100644
--- a/src/docs/man/hbase-args.txt
+++ b/src/docs/man/hbase-args.txt
@@ -36,4 +36,9 @@
 --hbase-table (table-name)::
   Specifies an HBase table to use as the target instead of HDFS
 
+--hbase-null-incremental-mode (mode)::
+  How to handle columns updated to null during incremental imports. +ignore+ is the default and
+  will result in retaining the previously imported value. +delete+ mode will delete all previous
+  versions of the column from HBase.
+
 
diff --git a/src/docs/user/hbase-args.txt b/src/docs/user/hbase-args.txt
index 53040f5..4076214 100644
--- a/src/docs/user/hbase-args.txt
+++ b/src/docs/user/hbase-args.txt
@@ -20,19 +20,22 @@
 
 .HBase arguments:
 [grid="all"]
-`-----------------------------`-------------------------------------------
-Argument                      Description
---------------------------------------------------------------------------
-+\--column-family <family>+   Sets the target column family for the import
-+\--hbase-create-table+       If specified, create missing HBase tables
-+\--hbase-row-key <col>+      Specifies which input column to use as the\
-                              row key
-                              In case, if input table contains composite
-                              key, then <col> must be in the form of a
-                              comma-separated list of composite key
-                              attributes
-+\--hbase-table <table-name>+ Specifies an HBase table to use as the \
-                              target instead of HDFS
-+\--hbase-bulkload+           Enables bulk loading
---------------------------------------------------------------------------
+`---------------------------------------`-------------------------------------------
+Argument                                Description
+------------------------------------------------------------------------------------
++\--column-family <family>+             Sets the target column family for the import
++\--hbase-create-table+                 If specified, create missing HBase tables
++\--hbase-row-key <col>+                Specifies which input column to use as the\
+                                        row key
+                                        In case, if input table contains composite
+                                        key, then <col> must be in the form of a
+                                        comma-separated list of composite key
+                                        attributes
++\--hbase-table <table-name>+           Specifies an HBase table to use as the \
+                                        target instead of HDFS
++\--hbase-bulkload+                     Enables bulk loading
++\--hbase-null-incremental-mode <mode>+ How to handle columns updated to null. \
+                                        Legal values for <mode> are +ignore+ \
+                                        (default) and +delete+.
+------------------------------------------------------------------------------------
 
diff --git a/src/docs/user/hbase.txt b/src/docs/user/hbase.txt
index ab4aedc..817956d 100644
--- a/src/docs/user/hbase.txt
+++ b/src/docs/user/hbase.txt
@@ -58,5 +58,10 @@
 cell. Sqoop will skip all rows containing null values in all columns
 except the row key column.
 
+By default Sqoop will retain the previously imported value for columns
+updated to null during incremental imports. This can be changed to
+delete all previous versions of the column by using
++\--hbase-null-incremental-mode delete+.
+
 To decrease the load on hbase, Sqoop can do bulk loading as opposed to
 direct writes. To use bulk loading, enable it using +\--hbase-bulkload+.
diff --git a/src/java/org/apache/sqoop/SqoopOptions.java b/src/java/org/apache/sqoop/SqoopOptions.java
index 73d0757..651cebd 100644
--- a/src/java/org/apache/sqoop/SqoopOptions.java
+++ b/src/java/org/apache/sqoop/SqoopOptions.java
@@ -105,6 +105,18 @@
   }
 
   /**
+   * How to handle null values when doing incremental import into HBase table:
+   * <ul>
+   * <li>Ignore: ignore update, retain previous value</li>
+   * <li>Delete: delete all previous values of column</li>
+   * </ul>
+   */
+  public enum HBaseNullIncrementalMode {
+    Ignore,
+    Delete,
+  }
+
+  /**
    * Update mode option specifies how updates are performed when
    * new rows are found with non-matching keys in database.
    * It supports two modes:
@@ -322,6 +334,9 @@
   @StoredAsProperty("incremental.last.value")
   private String incrementalLastValue;
 
+  @StoredAsProperty("hbase.null.incremental.mode")
+  private HBaseNullIncrementalMode hbaseNullIncrementalMode;
+
   // exclude these tables when importing all tables.
   @StoredAsProperty("import.all_tables.exclude")
   private String allTablesExclude;
@@ -1085,6 +1100,7 @@
     this.dbOutColumns = null;
 
     this.incrementalMode = IncrementalMode.None;
+    this.hbaseNullIncrementalMode = HBaseNullIncrementalMode.Ignore;
 
     this.updateMode = UpdateMode.UpdateOnly;
 
@@ -2301,6 +2317,20 @@
   }
 
   /**
+   * Get HBase null incremental mode to use.
+   */
+  public HBaseNullIncrementalMode getHbaseNullIncrementalMode() {
+    return hbaseNullIncrementalMode;
+  }
+
+  /**
+   * Set HBase null incremental mode to use.
+   */
+  public void setHbaseNullIncrementalMode(HBaseNullIncrementalMode hbaseNullIncrementalMode) {
+    this.hbaseNullIncrementalMode = hbaseNullIncrementalMode;
+  }
+
+  /**
    * Set the tables to be excluded when doing all table import.
    */
   public void setAllTablesExclude(String exclude) {
diff --git a/src/java/org/apache/sqoop/hbase/HBasePutProcessor.java b/src/java/org/apache/sqoop/hbase/HBasePutProcessor.java
index 27d6006..df9836b 100644
--- a/src/java/org/apache/sqoop/hbase/HBasePutProcessor.java
+++ b/src/java/org/apache/sqoop/hbase/HBasePutProcessor.java
@@ -63,6 +63,8 @@
   public static final String ROW_KEY_COLUMN_KEY =
       "sqoop.hbase.insert.row.key.column";
 
+  public static final String NULL_INCREMENTAL_MODE = "hbase.null.incremental.mode";
+
   /**
    * Configuration key specifying the PutTransformer implementation to use.
    */
diff --git a/src/java/org/apache/sqoop/hbase/ToStringPutTransformer.java b/src/java/org/apache/sqoop/hbase/ToStringPutTransformer.java
index 0bd6169..8600382 100644
--- a/src/java/org/apache/sqoop/hbase/ToStringPutTransformer.java
+++ b/src/java/org/apache/sqoop/hbase/ToStringPutTransformer.java
@@ -26,6 +26,7 @@
 import org.apache.hadoop.hbase.client.Put;
 import org.apache.hadoop.hbase.util.Bytes;
 import org.apache.hadoop.util.StringUtils;
+import org.apache.sqoop.SqoopOptions;
 import org.apache.sqoop.mapreduce.ImportJobBase;
 
 import java.io.IOException;
@@ -57,6 +58,7 @@
   protected boolean addRowKey;
   private boolean isCompositeKey = false;
   private List<String> compositeKeyAttributes;
+  private SqoopOptions.HBaseNullIncrementalMode nullMode;
 
   /**
    * Used as delimiter to combine composite-key column names when passed as.
@@ -170,6 +172,7 @@
     String colFamily, String rowKey) {
     byte[] colFamilyBytes = Bytes.toBytes(colFamily);
     List<Mutation> mutationList = new ArrayList<Mutation>();
+    Put put = null;
     for (Map.Entry<String, Object> fieldEntry : record.entrySet()) {
       String colName = fieldEntry.getKey();
       boolean rowKeyCol = false;
@@ -187,7 +190,10 @@
         Object val = fieldEntry.getValue();
         if (null != val) {
           // Put row-key in HBase
-          Put put = new Put(Bytes.toBytes(rowKey));
+          if (put == null) {
+            put = new Put(Bytes.toBytes(rowKey));
+            mutationList.add(put);
+          }
           if ( val instanceof byte[]) {
             put.addColumn(colFamilyBytes, getFieldNameBytes(colName),
                 (byte[])val);
@@ -197,9 +203,16 @@
           }
           mutationList.add(put);
         } else {
+          switch (nullMode) {
+          case Delete:
             Delete delete = new Delete(Bytes.toBytes(rowKey));
-            delete.addColumn(colFamilyBytes, getFieldNameBytes(colName));
+            delete.addColumns(colFamilyBytes, getFieldNameBytes(colName));
             mutationList.add(delete);
+            break;
+          case Ignore:
+            // Do nothing
+            break;
+          }
         }
       }
     }
@@ -218,6 +231,7 @@
 
   @Override
   public void init(Configuration conf) {
+    nullMode = conf.getEnum(HBasePutProcessor.NULL_INCREMENTAL_MODE, SqoopOptions.HBaseNullIncrementalMode.Ignore);
     setColumnFamily(conf.get(COL_FAMILY_KEY, null));
     setRowKeyColumn(conf.get(ROW_KEY_COLUMN_KEY, null));
 
diff --git a/src/java/org/apache/sqoop/mapreduce/HBaseImportJob.java b/src/java/org/apache/sqoop/mapreduce/HBaseImportJob.java
index 33da487..a09a45e 100644
--- a/src/java/org/apache/sqoop/mapreduce/HBaseImportJob.java
+++ b/src/java/org/apache/sqoop/mapreduce/HBaseImportJob.java
@@ -93,9 +93,10 @@
         HBasePutProcessor.class,
         FieldMapProcessor.class);
 
-    // Set the HBase parameters (table, column family, row key):
+    // Set the HBase parameters (table, column family, row key, null mode):
     conf.set(HBasePutProcessor.TABLE_NAME_KEY, options.getHBaseTable());
     conf.set(HBasePutProcessor.COL_FAMILY_KEY, options.getHBaseColFamily());
+    conf.set(HBasePutProcessor.NULL_INCREMENTAL_MODE, options.getHbaseNullIncrementalMode().toString());
 
     // What column of the input becomes the row key?
     String rowKeyCol = options.getHBaseRowKeyColumn();
diff --git a/src/java/org/apache/sqoop/tool/BaseSqoopTool.java b/src/java/org/apache/sqoop/tool/BaseSqoopTool.java
index ce21918..b02e4fe 100644
--- a/src/java/org/apache/sqoop/tool/BaseSqoopTool.java
+++ b/src/java/org/apache/sqoop/tool/BaseSqoopTool.java
@@ -199,6 +199,7 @@
   public static final String HBASE_BULK_LOAD_ENABLED_ARG =
       "hbase-bulkload";
   public static final String HBASE_CREATE_TABLE_ARG = "hbase-create-table";
+  public static final String HBASE_NULL_INCREMENTAL_MODE_ARG = "hbase-null-incremental-mode";
 
   //Accumulo arguments.
   public static final String ACCUMULO_TABLE_ARG = "accumulo-table";
@@ -853,6 +854,11 @@
         .withDescription("If specified, create missing HBase tables")
         .withLongOpt(HBASE_CREATE_TABLE_ARG)
         .create());
+    hbaseOpts.addOption(OptionBuilder.withArgName("nullmode")
+        .hasArg()
+        .withDescription("How to handle null values during incremental import into HBase.")
+        .withLongOpt(HBASE_NULL_INCREMENTAL_MODE_ARG)
+        .create());
 
     return hbaseOpts;
   }
@@ -1398,7 +1404,7 @@
     }
   }
 
-  protected void applyHBaseOptions(CommandLine in, SqoopOptions out) {
+  protected void applyHBaseOptions(CommandLine in, SqoopOptions out) throws InvalidOptionsException {
     if (in.hasOption(HBASE_TABLE_ARG)) {
       out.setHBaseTable(in.getOptionValue(HBASE_TABLE_ARG));
     }
@@ -1416,6 +1422,19 @@
     if (in.hasOption(HBASE_CREATE_TABLE_ARG)) {
       out.setCreateHBaseTable(true);
     }
+
+    if (in.hasOption(HBASE_NULL_INCREMENTAL_MODE_ARG)) {
+      String nullMode = in.getOptionValue(HBASE_NULL_INCREMENTAL_MODE_ARG);
+      if ("ignore".equals(nullMode)) {
+        out.setHbaseNullIncrementalMode(SqoopOptions.HBaseNullIncrementalMode.Ignore);
+      } else if ("delete".equals(nullMode)) {
+        out.setHbaseNullIncrementalMode(SqoopOptions.HBaseNullIncrementalMode.Delete);
+      } else {
+        throw new InvalidOptionsException("Unknown HBase null incremental mode: "
+            + nullMode + ". Use 'ignore' or 'delete'."
+            + HELP_STR);
+      }
+    }
   }
 
   protected void applyValidationOptions(CommandLine in, SqoopOptions out)
diff --git a/src/test/org/apache/sqoop/TestSqoopOptions.java b/src/test/org/apache/sqoop/TestSqoopOptions.java
index 16901ca..bb7c20d 100644
--- a/src/test/org/apache/sqoop/TestSqoopOptions.java
+++ b/src/test/org/apache/sqoop/TestSqoopOptions.java
@@ -89,6 +89,7 @@
     excludedFieldsFromClone.add("updateMode");
     excludedFieldsFromClone.add("layout");
     excludedFieldsFromClone.add("activeSqoopTool");
+    excludedFieldsFromClone.add("hbaseNullIncrementalMode");
   }
 
   @After
diff --git a/src/test/org/apache/sqoop/hbase/HBaseImportTest.java b/src/test/org/apache/sqoop/hbase/HBaseImportTest.java
index 2e73cf3..58c21ce 100644
--- a/src/test/org/apache/sqoop/hbase/HBaseImportTest.java
+++ b/src/test/org/apache/sqoop/hbase/HBaseImportTest.java
@@ -72,24 +72,6 @@
   }
 
   @Test
-  public void testOverwriteNullColumnsSucceeds() throws IOException {
-    // Test that we can create a table and then import immediately
-    // back on top of it without problem and then update with null to validate
-    String [] argv = getArgv(true, "OverwriteTable", "OverwriteColumnFamily", true, null);
-    String [] types = { "INT", "INT", "INT", "DATETIME" };
-    String [] vals = { "0", "1", "1", "'2017-03-20'" };
-    createTableWithColTypes(types, vals);
-    runImport(argv);
-    verifyHBaseCell("OverwriteTable", "0", "OverwriteColumnFamily", getColName(2), "1");
-    // Run a second time.
-    argv = getIncrementalArgv(true, "OverwriteTable", "OverwriteColumnFamily", true, null, false, false, "DATA_COL3", "2017-03-24 01:01:01.0", null);
-    vals = new String[] { "0", "1", null, "'2017-03-25'" };
-    updateTable(types, vals);
-    runImport(argv);
-    verifyHBaseCell("OverwriteTable", "0", "OverwriteColumnFamily", getColName(2), null);
-  }
-
-  @Test
   public void testAppendWithTimestampSucceeds() throws IOException {
     // Test that we can create a table and then import multiple rows
     // validate for append scenario with time stamp
@@ -100,7 +82,7 @@
     runImport(argv);
     verifyHBaseCell("AppendTable", "0", "AppendColumnFamily", getColName(2), "1");
     // Run a second time.
-    argv = getIncrementalArgv(true, "AppendTable", "AppendColumnFamily", true, null, true, false, "DATA_COL1", "2017-03-24 01:01:01.0", null);
+    argv = getIncrementalArgv(true, "AppendTable", "AppendColumnFamily", true, null, true, false, "DATA_COL1", "2017-03-24 01:01:01.0", null, "ignore");
     vals = new String[] { "1", "2", "3", "'2017-06-15'" };
     insertIntoTable(types, vals);
     runImport(argv);
@@ -118,7 +100,7 @@
     runImport(argv);
     verifyHBaseCell("AppendTable", "0", "AppendColumnFamily", getColName(2), "1");
     // Run a second time.
-    argv = getIncrementalArgv(true, "AppendTable", "AppendColumnFamily", true, null, true, true, "DATA_COL1", null, "DATA_COL3");
+    argv = getIncrementalArgv(true, "AppendTable", "AppendColumnFamily", true, null, true, true, "DATA_COL1", null, "DATA_COL3", "ignore");
     vals = new String[] { "1", "2", "3", "'2017-06-15'" };
     insertIntoTable(types, vals);
     runImport(argv);
@@ -126,6 +108,49 @@
   }
 
   @Test
+  public void testNullIncrementalModeIgnore() throws Exception {
+    // Latest value retained with 'ignore' mode
+    runInsertUpdateUpdateDeleteAndExpectValue("ignore", "2");
+  }
+
+  @Test
+  public void testNullIncrementalModeDelete() throws Exception {
+    // All previous values deleted with 'delete' mode
+    runInsertUpdateUpdateDeleteAndExpectValue("delete", null);
+  }
+
+  /**
+   * Does the following
+   *  - create HBase table
+   *  - insert value "1"
+   *  - update value to "2"
+   *  - update value to null
+   *  - asserts its value equals expectedValue
+   *
+   * @param nullMode hbase-null-incremental-mode to use ('ignore' or 'delete')
+   * @param expectedValue expected value in the end
+   * @throws Exception
+   */
+  private void runInsertUpdateUpdateDeleteAndExpectValue(String nullMode, String expectedValue) throws Exception {
+    // Create table and import with initial values
+    String [] types = { "INT", "INT", "DATETIME" };
+    createTableWithColTypes(types, new String[] { "0", "1", "'2017-03-20'" });
+    runImport(getArgv(true, "OverwriteTable", "OverwriteColumnFamily", true, null));
+    verifyHBaseCell("OverwriteTable", "0", "OverwriteColumnFamily", getColName(1), "1");
+
+    // Run a second time after updating.
+    updateTable(types, new String[] { "0", "2", "'2017-03-25'" });
+    runImport(getIncrementalArgv(true, "OverwriteTable", "OverwriteColumnFamily", true, null, false, false, getColName(2), "2017-03-24 01:01:01.0", null, nullMode));
+    verifyHBaseCell("OverwriteTable", "0", "OverwriteColumnFamily", getColName(1), "2");
+
+    // Run third time after deleting (setting to null)
+    updateTable(types, new String[] { "0", null, "'2017-03-28'" });
+    runImport(getIncrementalArgv(true, "OverwriteTable", "OverwriteColumnFamily", true, null, false, false, getColName(2), "2017-03-26 01:01:01.0", null, nullMode));
+    verifyHBaseCell("OverwriteTable", "0", "OverwriteColumnFamily", getColName(1), expectedValue);
+  }
+
+
+  @Test
   public void testExitFailure() throws IOException {
     String [] types = { "INT", "INT", "INT" };
     String [] vals = { "0", "42", "43" };
diff --git a/src/test/org/apache/sqoop/hbase/HBaseTestCase.java b/src/test/org/apache/sqoop/hbase/HBaseTestCase.java
index 98f8698..f96b658 100644
--- a/src/test/org/apache/sqoop/hbase/HBaseTestCase.java
+++ b/src/test/org/apache/sqoop/hbase/HBaseTestCase.java
@@ -138,7 +138,7 @@
    */
   protected String [] getIncrementalArgv(boolean includeHadoopFlags,
       String hbaseTable, String hbaseColFam, boolean hbaseCreate,
-      String queryStr, boolean isAppend, boolean appendTimestamp, String checkColumn, String checkValue, String lastModifiedColumn) {
+      String queryStr, boolean isAppend, boolean appendTimestamp, String checkColumn, String checkValue, String lastModifiedColumn, String nullMode) {
 
     String[] argsStrArray = getArgv(includeHadoopFlags, hbaseTable, hbaseColFam, hbaseCreate, queryStr);
     List<String> args = new ArrayList<String>(Arrays.asList(argsStrArray));
@@ -161,6 +161,14 @@
       args.add("--last-value");
       args.add(checkValue);
     }
+
+    // Set --hbase-null-incremental-mode (default is 'ignore')
+    if (nullMode == null) {
+      nullMode = "ignore";
+    }
+    args.add("--hbase-null-incremental-mode");
+    args.add(nullMode);
+
     return args.toArray(new String[0]);
   }