diff --git a/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/fs/FileSystem.java b/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/fs/FileSystem.java
index e18ea13cb34..59907bf7fc2 100644
--- a/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/fs/FileSystem.java
+++ b/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/fs/FileSystem.java
@@ -2644,6 +2644,16 @@ public abstract class FileSystem extends Configured implements Closeable {
         + " doesn't support setStoragePolicy");
   }
 
+  /**
+   * Unset the storage policy set for a given file or directory.
+   * @param src file or directory path.
+   * @throws IOException
+   */
+  public void unsetStoragePolicy(Path src) throws IOException {
+    throw new UnsupportedOperationException(getClass().getSimpleName()
+        + " doesn't support unsetStoragePolicy");
+  }
+
   /**
    * Query the effective storage policy ID for the given file or directory.
    *
diff --git a/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/fs/FilterFileSystem.java b/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/fs/FilterFileSystem.java
index 4ee75144727..3f9aaa4672a 100644
--- a/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/fs/FilterFileSystem.java
+++ b/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/fs/FilterFileSystem.java
@@ -632,6 +632,11 @@ public class FilterFileSystem extends FileSystem {
     fs.setStoragePolicy(src, policyName);
   }
 
+  @Override
+  public void unsetStoragePolicy(Path src) throws IOException {
+    fs.unsetStoragePolicy(src);
+  }
+
   @Override
   public BlockStoragePolicySpi getStoragePolicy(final Path src)
       throws IOException {
diff --git a/hadoop-common-project/hadoop-common/src/test/java/org/apache/hadoop/fs/TestHarFileSystem.java b/hadoop-common-project/hadoop-common/src/test/java/org/apache/hadoop/fs/TestHarFileSystem.java
index 26923a88172..a8795cc9c5c 100644
--- a/hadoop-common-project/hadoop-common/src/test/java/org/apache/hadoop/fs/TestHarFileSystem.java
+++ b/hadoop-common-project/hadoop-common/src/test/java/org/apache/hadoop/fs/TestHarFileSystem.java
@@ -210,6 +210,8 @@ public class TestHarFileSystem {
     public void setStoragePolicy(Path src, String policyName)
         throws IOException;
 
+    public void unsetStoragePolicy(Path src) throws IOException;
+
     public BlockStoragePolicySpi getStoragePolicy(final Path src)
         throws IOException;
 
diff --git a/hadoop-hdfs-project/hadoop-hdfs-client/src/main/java/org/apache/hadoop/hdfs/DFSClient.java b/hadoop-hdfs-project/hadoop-hdfs-client/src/main/java/org/apache/hadoop/hdfs/DFSClient.java
index b87fec3fdb9..347f1a0ab9a 100644
--- a/hadoop-hdfs-project/hadoop-hdfs-client/src/main/java/org/apache/hadoop/hdfs/DFSClient.java
+++ b/hadoop-hdfs-project/hadoop-hdfs-client/src/main/java/org/apache/hadoop/hdfs/DFSClient.java
@@ -1501,6 +1501,24 @@ public class DFSClient implements java.io.Closeable, RemotePeerFactory,
     }
   }
 
+  /**
+   * Unset storage policy set for a given file/directory.
+   * @param src file/directory name
+   */
+  public void unsetStoragePolicy(String src) throws IOException {
+    checkOpen();
+    try (TraceScope ignored = newPathTraceScope("unsetStoragePolicy", src)) {
+      namenode.unsetStoragePolicy(src);
+    } catch (RemoteException e) {
+      throw e.unwrapRemoteException(AccessControlException.class,
+          FileNotFoundException.class,
+          SafeModeException.class,
+          NSQuotaExceededException.class,
+          UnresolvedPathException.class,
+          SnapshotAccessControlException.class);
+    }
+  }
+
   /**
    * @param path file/directory name
    * @return Get the storage policy for specified path
diff --git a/hadoop-hdfs-project/hadoop-hdfs-client/src/main/java/org/apache/hadoop/hdfs/DistributedFileSystem.java b/hadoop-hdfs-project/hadoop-hdfs-client/src/main/java/org/apache/hadoop/hdfs/DistributedFileSystem.java
index 5bca8a28e4f..ae14d468153 100644
--- a/hadoop-hdfs-project/hadoop-hdfs-client/src/main/java/org/apache/hadoop/hdfs/DistributedFileSystem.java
+++ b/hadoop-hdfs-project/hadoop-hdfs-client/src/main/java/org/apache/hadoop/hdfs/DistributedFileSystem.java
@@ -549,6 +549,31 @@ public class DistributedFileSystem extends FileSystem {
     }.resolve(this, absF);
   }
 
+  @Override
+  public void unsetStoragePolicy(final Path src)
+      throws IOException {
+    statistics.incrementWriteOps(1);
+    Path absF = fixRelativePart(src);
+    new FileSystemLinkResolver<Void>() {
+      @Override
+      public Void doCall(final Path p) throws IOException {
+        dfs.unsetStoragePolicy(getPathName(p));
+        return null;
+      }
+      @Override
+      public Void next(final FileSystem fs, final Path p) throws IOException {
+        if (fs instanceof DistributedFileSystem) {
+          ((DistributedFileSystem) fs).unsetStoragePolicy(p);
+          return null;
+        } else {
+          throw new UnsupportedOperationException(
+              "Cannot perform unsetStoragePolicy on a "
+                  + "non-DistributedFileSystem: " + src + " -> " + p);
+        }
+      }
+    }.resolve(this, absF);
+  }
+
   @Override
   public BlockStoragePolicySpi getStoragePolicy(Path path) throws IOException {
     statistics.incrementReadOps(1);
diff --git a/hadoop-hdfs-project/hadoop-hdfs-client/src/main/java/org/apache/hadoop/hdfs/protocol/ClientProtocol.java b/hadoop-hdfs-project/hadoop-hdfs-client/src/main/java/org/apache/hadoop/hdfs/protocol/ClientProtocol.java
index 5fa80257699..93d409d0927 100644
--- a/hadoop-hdfs-project/hadoop-hdfs-client/src/main/java/org/apache/hadoop/hdfs/protocol/ClientProtocol.java
+++ b/hadoop-hdfs-project/hadoop-hdfs-client/src/main/java/org/apache/hadoop/hdfs/protocol/ClientProtocol.java
@@ -281,6 +281,19 @@ public interface ClientProtocol {
   void setStoragePolicy(String src, String policyName)
       throws IOException;
 
+  /**
+   * Unset the storage policy set for a given file or directory.
+   * @param src Path of an existing file/directory.
+   * @throws SnapshotAccessControlException If access is denied
+   * @throws org.apache.hadoop.fs.UnresolvedLinkException if <code>src</code>
+   *           contains a symlink
+   * @throws java.io.FileNotFoundException If file/dir <code>src</code> is not
+   *           found
+   * @throws QuotaExceededException If changes violate the quota restriction
+   */
+  @Idempotent
+  void unsetStoragePolicy(String src) throws IOException;
+
   /**
    * Get the storage policy for a file/directory.
    * @param path
diff --git a/hadoop-hdfs-project/hadoop-hdfs-client/src/main/java/org/apache/hadoop/hdfs/protocolPB/ClientNamenodeProtocolTranslatorPB.java b/hadoop-hdfs-project/hadoop-hdfs-client/src/main/java/org/apache/hadoop/hdfs/protocolPB/ClientNamenodeProtocolTranslatorPB.java
index c94d5151c23..8844f2aa1b2 100644
--- a/hadoop-hdfs-project/hadoop-hdfs-client/src/main/java/org/apache/hadoop/hdfs/protocolPB/ClientNamenodeProtocolTranslatorPB.java
+++ b/hadoop-hdfs-project/hadoop-hdfs-client/src/main/java/org/apache/hadoop/hdfs/protocolPB/ClientNamenodeProtocolTranslatorPB.java
@@ -134,6 +134,7 @@ import org.apache.hadoop.hdfs.protocol.proto.ClientNamenodeProtocolProtos.Recove
 import org.apache.hadoop.hdfs.protocol.proto.ClientNamenodeProtocolProtos.RefreshNodesRequestProto;
 import org.apache.hadoop.hdfs.protocol.proto.ClientNamenodeProtocolProtos.RemoveCacheDirectiveRequestProto;
 import org.apache.hadoop.hdfs.protocol.proto.ClientNamenodeProtocolProtos.RemoveCachePoolRequestProto;
+import org.apache.hadoop.hdfs.protocol.proto.ClientNamenodeProtocolProtos.UnsetStoragePolicyRequestProto;
 import org.apache.hadoop.hdfs.protocol.proto.ClientNamenodeProtocolProtos.Rename2RequestProto;
 import org.apache.hadoop.hdfs.protocol.proto.ClientNamenodeProtocolProtos.RenameRequestProto;
 import org.apache.hadoop.hdfs.protocol.proto.ClientNamenodeProtocolProtos.RenameSnapshotRequestProto;
@@ -1438,6 +1439,17 @@ public class ClientNamenodeProtocolTranslatorPB implements
     }
   }
 
+  @Override
+  public void unsetStoragePolicy(String src) throws IOException {
+    UnsetStoragePolicyRequestProto req = UnsetStoragePolicyRequestProto
+        .newBuilder().setSrc(src).build();
+    try {
+      rpcProxy.unsetStoragePolicy(null, req);
+    } catch (ServiceException e) {
+      throw ProtobufHelper.getRemoteException(e);
+    }
+  }
+
   @Override
   public BlockStoragePolicy getStoragePolicy(String path) throws IOException {
     GetStoragePolicyRequestProto request = GetStoragePolicyRequestProto
diff --git a/hadoop-hdfs-project/hadoop-hdfs-client/src/main/proto/ClientNamenodeProtocol.proto b/hadoop-hdfs-project/hadoop-hdfs-client/src/main/proto/ClientNamenodeProtocol.proto
index 9925d1d3bbe..87a76f25ba7 100644
--- a/hadoop-hdfs-project/hadoop-hdfs-client/src/main/proto/ClientNamenodeProtocol.proto
+++ b/hadoop-hdfs-project/hadoop-hdfs-client/src/main/proto/ClientNamenodeProtocol.proto
@@ -112,6 +112,13 @@ message SetStoragePolicyRequestProto {
 message SetStoragePolicyResponseProto { // void response
 }
 
+message UnsetStoragePolicyRequestProto {
+  required string src = 1;
+}
+
+message UnsetStoragePolicyResponseProto {
+}
+
 message GetStoragePolicyRequestProto {
   required string path = 1;
 }
@@ -741,6 +748,8 @@ service ClientNamenodeProtocol {
       returns(SetReplicationResponseProto);
   rpc setStoragePolicy(SetStoragePolicyRequestProto)
       returns(SetStoragePolicyResponseProto);
+  rpc unsetStoragePolicy(UnsetStoragePolicyRequestProto)
+      returns(UnsetStoragePolicyResponseProto);
   rpc getStoragePolicy(GetStoragePolicyRequestProto)
       returns(GetStoragePolicyResponseProto);
   rpc getStoragePolicies(GetStoragePoliciesRequestProto)
diff --git a/hadoop-hdfs-project/hadoop-hdfs/CHANGES.txt b/hadoop-hdfs-project/hadoop-hdfs/CHANGES.txt
index 68986292538..759bb133f38 100644
--- a/hadoop-hdfs-project/hadoop-hdfs/CHANGES.txt
+++ b/hadoop-hdfs-project/hadoop-hdfs/CHANGES.txt
@@ -74,6 +74,9 @@ Release 2.9.0 - UNRELEASED
 
     HDFS-7964. Add support for async edit logging. (Daryn Sharp)
 
+    HDFS-9534. Add CLI command to clear storage policy from a path.
+    (Xiaobing Zhou via Arpit Agarwal)
+
   OPTIMIZATIONS
 
   BUG FIXES
diff --git a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/protocolPB/ClientNamenodeProtocolServerSideTranslatorPB.java b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/protocolPB/ClientNamenodeProtocolServerSideTranslatorPB.java
index a88307f8a9c..2c19db9e0bd 100644
--- a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/protocolPB/ClientNamenodeProtocolServerSideTranslatorPB.java
+++ b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/protocolPB/ClientNamenodeProtocolServerSideTranslatorPB.java
@@ -157,6 +157,8 @@ import org.apache.hadoop.hdfs.protocol.proto.ClientNamenodeProtocolProtos.Remove
 import org.apache.hadoop.hdfs.protocol.proto.ClientNamenodeProtocolProtos.RemoveCacheDirectiveResponseProto;
 import org.apache.hadoop.hdfs.protocol.proto.ClientNamenodeProtocolProtos.RemoveCachePoolRequestProto;
 import org.apache.hadoop.hdfs.protocol.proto.ClientNamenodeProtocolProtos.RemoveCachePoolResponseProto;
+import org.apache.hadoop.hdfs.protocol.proto.ClientNamenodeProtocolProtos.UnsetStoragePolicyRequestProto;
+import org.apache.hadoop.hdfs.protocol.proto.ClientNamenodeProtocolProtos.UnsetStoragePolicyResponseProto;
 import org.apache.hadoop.hdfs.protocol.proto.ClientNamenodeProtocolProtos.Rename2RequestProto;
 import org.apache.hadoop.hdfs.protocol.proto.ClientNamenodeProtocolProtos.Rename2ResponseProto;
 import org.apache.hadoop.hdfs.protocol.proto.ClientNamenodeProtocolProtos.RenameRequestProto;
@@ -255,6 +257,9 @@ public class ClientNamenodeProtocolServerSideTranslatorPB implements
       GetSnapshottableDirListingResponseProto.newBuilder().build();
   static final SetStoragePolicyResponseProto VOID_SET_STORAGE_POLICY_RESPONSE =
       SetStoragePolicyResponseProto.newBuilder().build();
+  static final UnsetStoragePolicyResponseProto
+      VOID_UNSET_STORAGE_POLICY_RESPONSE =
+      UnsetStoragePolicyResponseProto.newBuilder().build();
 
   private static final CreateResponseProto VOID_CREATE_RESPONSE = 
   CreateResponseProto.newBuilder().build();
@@ -1465,6 +1470,18 @@ public class ClientNamenodeProtocolServerSideTranslatorPB implements
     return VOID_SET_STORAGE_POLICY_RESPONSE;
   }
 
+  @Override
+  public UnsetStoragePolicyResponseProto unsetStoragePolicy(
+      RpcController controller, UnsetStoragePolicyRequestProto request)
+      throws ServiceException {
+    try {
+      server.unsetStoragePolicy(request.getSrc());
+    } catch (IOException e) {
+      throw new ServiceException(e);
+    }
+    return VOID_UNSET_STORAGE_POLICY_RESPONSE;
+  }
+
   @Override
   public GetStoragePolicyResponseProto getStoragePolicy(
       RpcController controller, GetStoragePolicyRequestProto request)
diff --git a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/namenode/FSDirAttrOp.java b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/namenode/FSDirAttrOp.java
index 7f4107827b6..7575f5ec339 100644
--- a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/namenode/FSDirAttrOp.java
+++ b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/namenode/FSDirAttrOp.java
@@ -37,6 +37,8 @@ import org.apache.hadoop.hdfs.server.blockmanagement.BlockStoragePolicySuite;
 import org.apache.hadoop.hdfs.util.EnumCounters;
 import org.apache.hadoop.security.AccessControlException;
 
+import com.google.common.collect.Lists;
+
 import java.io.FileNotFoundException;
 import java.io.IOException;
 import java.util.Arrays;
@@ -158,13 +160,31 @@ public class FSDirAttrOp {
     return isFile;
   }
 
-  static HdfsFileStatus setStoragePolicy(
-      FSDirectory fsd, BlockManager bm, String src, final String policyName)
+  static HdfsFileStatus unsetStoragePolicy(FSDirectory fsd, BlockManager bm,
+      String src) throws IOException {
+    return setStoragePolicy(fsd, bm, src,
+        HdfsConstants.BLOCK_STORAGE_POLICY_ID_UNSPECIFIED, "unset");
+  }
+
+  static HdfsFileStatus setStoragePolicy(FSDirectory fsd, BlockManager bm,
+      String src, final String policyName) throws IOException {
+    // get the corresponding policy and make sure the policy name is valid
+    BlockStoragePolicy policy = bm.getStoragePolicy(policyName);
+    if (policy == null) {
+      throw new HadoopIllegalArgumentException(
+          "Cannot find a block policy with the name " + policyName);
+    }
+
+    return setStoragePolicy(fsd, bm, src, policy.getId(), "set");
+  }
+
+  static HdfsFileStatus setStoragePolicy(FSDirectory fsd, BlockManager bm,
+      String src, final byte policyId, final String operation)
       throws IOException {
     if (!fsd.isStoragePolicyEnabled()) {
-      throw new IOException(
-          "Failed to set storage policy since "
-              + DFS_STORAGE_POLICY_ENABLED_KEY + " is set to false.");
+      throw new IOException(String.format(
+          "Failed to %s storage policy since %s is set to false.", operation,
+          DFS_STORAGE_POLICY_ENABLED_KEY));
     }
     FSPermissionChecker pc = fsd.getPermissionChecker();
     byte[][] pathComponents = FSDirectory.getPathComponentsForReservedPath(src);
@@ -178,14 +198,8 @@ public class FSDirAttrOp {
         fsd.checkPathAccess(pc, iip, FsAction.WRITE);
       }
 
-      // get the corresponding policy and make sure the policy name is valid
-      BlockStoragePolicy policy = bm.getStoragePolicy(policyName);
-      if (policy == null) {
-        throw new HadoopIllegalArgumentException(
-            "Cannot find a block policy with the name " + policyName);
-      }
-      unprotectedSetStoragePolicy(fsd, bm, iip, policy.getId());
-      fsd.getEditLog().logSetStoragePolicy(src, policy.getId());
+      unprotectedSetStoragePolicy(fsd, bm, iip, policyId);
+      fsd.getEditLog().logSetStoragePolicy(src, policyId);
     } finally {
       fsd.writeUnlock();
     }
@@ -445,8 +459,8 @@ public class FSDirAttrOp {
     return file.getBlocks();
   }
 
-  static void unprotectedSetStoragePolicy(
-      FSDirectory fsd, BlockManager bm, INodesInPath iip, byte policyId)
+  static void unprotectedSetStoragePolicy(FSDirectory fsd, BlockManager bm,
+      INodesInPath iip, final byte policyId)
       throws IOException {
     assert fsd.hasWriteLock();
     final INode inode = iip.getLastINode();
@@ -456,10 +470,12 @@ public class FSDirAttrOp {
     }
     final int snapshotId = iip.getLatestSnapshotId();
     if (inode.isFile()) {
-      BlockStoragePolicy newPolicy = bm.getStoragePolicy(policyId);
-      if (newPolicy.isCopyOnCreateFile()) {
-        throw new HadoopIllegalArgumentException(
-            "Policy " + newPolicy + " cannot be set after file creation.");
+      if (policyId != HdfsConstants.BLOCK_STORAGE_POLICY_ID_UNSPECIFIED) {
+        BlockStoragePolicy newPolicy = bm.getStoragePolicy(policyId);
+        if (newPolicy.isCopyOnCreateFile()) {
+          throw new HadoopIllegalArgumentException("Policy " + newPolicy
+              + " cannot be set after file creation.");
+        }
       }
 
       BlockStoragePolicy currentPolicy =
@@ -472,7 +488,8 @@ public class FSDirAttrOp {
       }
       inode.asFile().setStoragePolicyID(policyId, snapshotId);
     } else if (inode.isDirectory()) {
-      setDirStoragePolicy(fsd, inode.asDirectory(), policyId, snapshotId);
+      setDirStoragePolicy(fsd, inode.asDirectory(), policyId,
+          snapshotId);
     } else {
       throw new FileNotFoundException(iip.getPath()
           + " is not a file or directory");
@@ -484,11 +501,18 @@ public class FSDirAttrOp {
       int latestSnapshotId) throws IOException {
     List<XAttr> existingXAttrs = XAttrStorage.readINodeXAttrs(inode);
     XAttr xAttr = BlockStoragePolicySuite.buildXAttr(policyId);
-    List<XAttr> newXAttrs = FSDirXAttrOp.setINodeXAttrs(fsd, existingXAttrs,
-                                                        Arrays.asList(xAttr),
-                                                        EnumSet.of(
-                                                            XAttrSetFlag.CREATE,
-                                                            XAttrSetFlag.REPLACE));
+    List<XAttr> newXAttrs = null;
+    if (policyId == HdfsConstants.BLOCK_STORAGE_POLICY_ID_UNSPECIFIED) {
+      List<XAttr> toRemove = Lists.newArrayList();
+      toRemove.add(xAttr);
+      List<XAttr> removed = Lists.newArrayList();
+      newXAttrs = FSDirXAttrOp.filterINodeXAttrs(existingXAttrs, toRemove,
+          removed);
+    } else {
+      newXAttrs = FSDirXAttrOp.setINodeXAttrs(fsd, existingXAttrs,
+          Arrays.asList(xAttr),
+          EnumSet.of(XAttrSetFlag.CREATE, XAttrSetFlag.REPLACE));
+    }
     XAttrStorage.updateINodeXAttrs(inode, newXAttrs, latestSnapshotId);
   }
 
diff --git a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/namenode/FSNamesystem.java b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/namenode/FSNamesystem.java
index 7fa977a3c0a..dbf360dbc94 100644
--- a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/namenode/FSNamesystem.java
+++ b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/namenode/FSNamesystem.java
@@ -1942,6 +1942,28 @@ public class FSNamesystem implements Namesystem, FSNamesystemMBean,
     logAuditEvent(true, "setStoragePolicy", src, null, auditStat);
   }
 
+  /**
+   * unset storage policy set for a given file or a directory.
+   *
+   * @param src file/directory path
+   */
+  void unsetStoragePolicy(String src) throws IOException {
+    HdfsFileStatus auditStat;
+    checkOperation(OperationCategory.WRITE);
+    writeLock();
+    try {
+      checkOperation(OperationCategory.WRITE);
+      checkNameNodeSafeMode("Cannot unset storage policy for " + src);
+      auditStat = FSDirAttrOp.unsetStoragePolicy(dir, blockManager, src);
+    } catch (AccessControlException e) {
+      logAuditEvent(false, "unsetStoragePolicy", src);
+      throw e;
+    } finally {
+      writeUnlock();
+    }
+    getEditLog().logSync();
+    logAuditEvent(true, "unsetStoragePolicy", src, null, auditStat);
+  }
   /**
    * Get the storage policy for a file or a directory.
    *
diff --git a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/namenode/NameNodeRpcServer.java b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/namenode/NameNodeRpcServer.java
index 1b4ef9c275b..554ea101123 100644
--- a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/namenode/NameNodeRpcServer.java
+++ b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/namenode/NameNodeRpcServer.java
@@ -765,6 +765,13 @@ class NameNodeRpcServer implements NamenodeProtocols {
     return namesystem.setReplication(src, replication);
   }
 
+  @Override
+  public void unsetStoragePolicy(String src)
+      throws IOException {
+    checkNNStartup();
+    namesystem.unsetStoragePolicy(src);
+  }
+
   @Override
   public void setStoragePolicy(String src, String policyName)
       throws IOException {
diff --git a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/tools/StoragePolicyAdmin.java b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/tools/StoragePolicyAdmin.java
index 98c8a6b8d3f..24079b3c25d 100644
--- a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/tools/StoragePolicyAdmin.java
+++ b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/tools/StoragePolicyAdmin.java
@@ -224,9 +224,55 @@ public class StoragePolicyAdmin extends Configured implements Tool {
     }
   }
 
+  /* Command to unset the storage policy set for a file/directory */
+  private static class UnsetStoragePolicyCommand
+      implements AdminHelper.Command {
+
+    @Override
+    public String getName() {
+      return "-unsetStoragePolicy";
+    }
+
+    @Override
+    public String getShortUsage() {
+      return "[" + getName() + " -path <path>]\n";
+    }
+
+    @Override
+    public String getLongUsage() {
+      TableListing listing = AdminHelper.getOptionDescriptionListing();
+      listing.addRow("<path>", "The path of the file/directory "
+          + "from which the storage policy will be unset.");
+      return getShortUsage() + "\n"
+          + "Unset the storage policy set for a file/directory.\n\n"
+          + listing.toString();
+    }
+
+    @Override
+    public int run(Configuration conf, List<String> args) throws IOException {
+      final String path = StringUtils.popOptionWithArgument("-path", args);
+      if (path == null) {
+        System.err.println("Please specify the path from which "
+            + "the storage policy will be unsetd.\nUsage: " + getLongUsage());
+        return 1;
+      }
+
+      final DistributedFileSystem dfs = AdminHelper.getDFS(conf);
+      try {
+        dfs.unsetStoragePolicy(new Path(path));
+        System.out.println("Unset storage policy from " + path);
+      } catch (Exception e) {
+        System.err.println(AdminHelper.prettifyException(e));
+        return 2;
+      }
+      return 0;
+    }
+  }
+
   private static final AdminHelper.Command[] COMMANDS = {
       new ListStoragePoliciesCommand(),
       new SetStoragePolicyCommand(),
-      new GetStoragePolicyCommand()
+      new GetStoragePolicyCommand(),
+      new UnsetStoragePolicyCommand()
   };
 }
\ No newline at end of file
diff --git a/hadoop-hdfs-project/hadoop-hdfs/src/site/markdown/ArchivalStorage.md b/hadoop-hdfs-project/hadoop-hdfs/src/site/markdown/ArchivalStorage.md
index cbfbaa16d5b..803cc9171d1 100644
--- a/hadoop-hdfs-project/hadoop-hdfs/src/site/markdown/ArchivalStorage.md
+++ b/hadoop-hdfs-project/hadoop-hdfs/src/site/markdown/ArchivalStorage.md
@@ -26,6 +26,7 @@ Archival Storage, SSD & Memory
     * [Storage Policy Commands](#Storage_Policy_Commands)
         * [List Storage Policies](#List_Storage_Policies)
         * [Set Storage Policy](#Set_Storage_Policy)
+        * [Unset Storage Policy](#Unset_Storage_Policy)
         * [Get Storage Policy](#Get_Storage_Policy)
 
 Introduction
@@ -150,6 +151,20 @@ Set a storage policy to a file or a directory.
 | `-path <path>` | The path referring to either a directory or a file. |
 | `-policy <policy>` | The name of the storage policy. |
 
+### Unset Storage Policy
+
+Unset a storage policy to a file or a directory. After the unset command the storage policy of the nearest ancestor will apply, and if there is no policy on any ancestor then the default storage policy will apply.
+
+* Command:
+
+        hdfs storagepolicies -unsetStoragePolicy -path <path>
+
+* Arguments:
+
+| | |
+|:---- |:---- |
+| `-path <path>` | The path referring to either a directory or a file. |
+
 ### Get Storage Policy
 
 Get the storage policy of a file or a directory.
diff --git a/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/TestApplyingStoragePolicy.java b/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/TestApplyingStoragePolicy.java
new file mode 100644
index 00000000000..200fab6f8aa
--- /dev/null
+++ b/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/TestApplyingStoragePolicy.java
@@ -0,0 +1,268 @@
+/**
+ * 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.hadoop.hdfs;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertTrue;
+
+import java.io.FileNotFoundException;
+import java.io.IOException;
+
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.Path;
+import org.apache.hadoop.hdfs.protocol.BlockStoragePolicy;
+import org.apache.hadoop.hdfs.server.blockmanagement.BlockStoragePolicySuite;
+import org.junit.After;
+import org.junit.Before;
+import org.junit.Test;
+
+public class TestApplyingStoragePolicy {
+  private static final short REPL = 1;
+  private static final int SIZE = 128;
+
+  private static Configuration conf;
+  private static MiniDFSCluster cluster;
+  private static DistributedFileSystem fs;
+
+  @Before
+  public void clusterSetUp() throws IOException {
+    conf = new HdfsConfiguration();
+    cluster = new MiniDFSCluster.Builder(conf).numDataNodes(REPL).build();
+    cluster.waitActive();
+    fs = cluster.getFileSystem();
+  }
+
+  @After
+  public void clusterShutdown() throws IOException{
+    if(fs != null) {
+      fs.close();
+      fs = null;
+    }
+    if(cluster != null) {
+      cluster.shutdown();
+      cluster = null;
+    }
+  }
+
+  @Test
+  public void testStoragePolicyByDefault() throws Exception {
+    final Path foo = new Path("/foo");
+    final Path bar = new Path(foo, "bar");
+    final Path wow = new Path(bar, "wow");
+    final Path fooz = new Path(bar, "/fooz");
+    DFSTestUtil.createFile(fs, wow, SIZE, REPL, 0);
+
+    final BlockStoragePolicySuite suite = BlockStoragePolicySuite
+        .createDefaultSuite();
+    final BlockStoragePolicy hot = suite.getPolicy("HOT");
+
+    /*
+     * test: storage policy is HOT by default or inherited from nearest
+     * ancestor, if not explicitly specified for newly created dir/file.
+     */
+    assertEquals(fs.getStoragePolicy(foo), hot);
+    assertEquals(fs.getStoragePolicy(bar), hot);
+    assertEquals(fs.getStoragePolicy(wow), hot);
+    try {
+      fs.getStoragePolicy(fooz);
+    } catch (Exception e) {
+      assertTrue(e instanceof FileNotFoundException);
+    }
+  }
+
+  @Test
+  public void testSetAndUnsetStoragePolicy() throws Exception {
+    final Path foo = new Path("/foo");
+    final Path bar = new Path(foo, "bar");
+    final Path wow = new Path(bar, "wow");
+    final Path fooz = new Path(bar, "/fooz");
+    DFSTestUtil.createFile(fs, wow, SIZE, REPL, 0);
+
+    final BlockStoragePolicySuite suite = BlockStoragePolicySuite
+        .createDefaultSuite();
+    final BlockStoragePolicy warm = suite.getPolicy("WARM");
+    final BlockStoragePolicy cold = suite.getPolicy("COLD");
+    final BlockStoragePolicy hot = suite.getPolicy("HOT");
+
+    /*
+     * test: set storage policy
+     */
+    fs.setStoragePolicy(foo, warm.getName());
+    fs.setStoragePolicy(bar, cold.getName());
+    fs.setStoragePolicy(wow, hot.getName());
+    try {
+      fs.setStoragePolicy(fooz, warm.getName());
+    } catch (Exception e) {
+      assertTrue(e instanceof FileNotFoundException);
+    }
+
+    /*
+     * test: get storage policy after set
+     */
+    assertEquals(fs.getStoragePolicy(foo), warm);
+    assertEquals(fs.getStoragePolicy(bar), cold);
+    assertEquals(fs.getStoragePolicy(wow), hot);
+    try {
+      fs.getStoragePolicy(fooz);
+    } catch (Exception e) {
+      assertTrue(e instanceof FileNotFoundException);
+    }
+
+    /*
+     * test: unset storage policy
+     */
+    fs.unsetStoragePolicy(foo);
+    fs.unsetStoragePolicy(bar);
+    fs.unsetStoragePolicy(wow);
+    try {
+      fs.unsetStoragePolicy(fooz);
+    } catch (Exception e) {
+      assertTrue(e instanceof FileNotFoundException);
+    }
+
+    /*
+     * test: get storage policy after unset
+     */
+    assertEquals(fs.getStoragePolicy(foo), hot);
+    assertEquals(fs.getStoragePolicy(bar), hot);
+    assertEquals(fs.getStoragePolicy(wow), hot);
+    try {
+      fs.getStoragePolicy(fooz);
+    } catch (Exception e) {
+      assertTrue(e instanceof FileNotFoundException);
+    }
+  }
+
+  @Test
+  public void testNestedStoragePolicy() throws Exception {
+    final Path foo = new Path("/foo");
+    final Path bar = new Path(foo, "bar");
+    final Path wow = new Path(bar, "wow");
+    final Path fooz = new Path("/foos");
+    DFSTestUtil.createFile(fs, wow, SIZE, REPL, 0);
+
+    final BlockStoragePolicySuite suite = BlockStoragePolicySuite
+        .createDefaultSuite();
+    final BlockStoragePolicy warm = suite.getPolicy("WARM");
+    final BlockStoragePolicy cold = suite.getPolicy("COLD");
+    final BlockStoragePolicy hot = suite.getPolicy("HOT");
+
+    /*
+     * test: set storage policy
+     */
+    fs.setStoragePolicy(foo, warm.getName());
+    fs.setStoragePolicy(bar, cold.getName());
+    fs.setStoragePolicy(wow, hot.getName());
+    try {
+      fs.setStoragePolicy(fooz, warm.getName());
+    } catch (Exception e) {
+      assertTrue(e instanceof FileNotFoundException);
+    }
+
+    /*
+     * test: get storage policy after set
+     */
+    assertEquals(fs.getStoragePolicy(foo), warm);
+    assertEquals(fs.getStoragePolicy(bar), cold);
+    assertEquals(fs.getStoragePolicy(wow), hot);
+    try {
+      fs.getStoragePolicy(fooz);
+    } catch (Exception e) {
+      assertTrue(e instanceof FileNotFoundException);
+    }
+
+    /*
+     * test: unset storage policy in the case of being nested
+     */
+    // unset wow
+    fs.unsetStoragePolicy(wow);
+    // inherit storage policy from wow's nearest ancestor
+    assertEquals(fs.getStoragePolicy(wow), cold);
+    // unset bar
+    fs.unsetStoragePolicy(bar);
+    // inherit storage policy from bar's nearest ancestor
+    assertEquals(fs.getStoragePolicy(bar), warm);
+    // unset foo
+    fs.unsetStoragePolicy(foo);
+    // default storage policy is applied, since no more available ancestors
+    assertEquals(fs.getStoragePolicy(foo), hot);
+    // unset fooz
+    try {
+      fs.unsetStoragePolicy(fooz);
+    } catch (Exception e) {
+      assertTrue(e instanceof FileNotFoundException);
+    }
+
+    /*
+     * test: default storage policy is applied, since no explicit policies from
+     * ancestors are available
+     */
+    assertEquals(fs.getStoragePolicy(foo), hot);
+    assertEquals(fs.getStoragePolicy(bar), hot);
+    assertEquals(fs.getStoragePolicy(wow), hot);
+    try {
+      fs.getStoragePolicy(fooz);
+    } catch (Exception e) {
+      assertTrue(e instanceof FileNotFoundException);
+    }
+  }
+
+  @Test
+  public void testSetAndGetStoragePolicy() throws IOException {
+    final Path foo = new Path("/foo");
+    final Path bar = new Path(foo, "bar");
+    final Path fooz = new Path("/fooz");
+    DFSTestUtil.createFile(fs, bar, SIZE, REPL, 0);
+
+    final BlockStoragePolicySuite suite = BlockStoragePolicySuite
+        .createDefaultSuite();
+    final BlockStoragePolicy warm = suite.getPolicy("WARM");
+    final BlockStoragePolicy cold = suite.getPolicy("COLD");
+    final BlockStoragePolicy hot = suite.getPolicy("HOT");
+
+    assertEquals(fs.getStoragePolicy(foo), hot);
+    assertEquals(fs.getStoragePolicy(bar), hot);
+    try {
+      fs.getStoragePolicy(fooz);
+    } catch (Exception e) {
+      assertTrue(e instanceof FileNotFoundException);
+    }
+
+    /*
+     * test: set storage policy
+     */
+    fs.setStoragePolicy(foo, warm.getName());
+    fs.setStoragePolicy(bar, cold.getName());
+    try {
+      fs.setStoragePolicy(fooz, warm.getName());
+    } catch (Exception e) {
+      assertTrue(e instanceof FileNotFoundException);
+    }
+
+    /*
+     * test: get storage policy after set
+     */
+    assertEquals(fs.getStoragePolicy(foo), warm);
+    assertEquals(fs.getStoragePolicy(bar), cold);
+    try {
+      fs.getStoragePolicy(fooz);
+    } catch (Exception e) {
+      assertTrue(e instanceof FileNotFoundException);
+    }
+  }
+}
diff --git a/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/tools/TestStoragePolicyCommands.java b/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/tools/TestStoragePolicyCommands.java
index ec0bb665de4..63f8484d76b 100644
--- a/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/tools/TestStoragePolicyCommands.java
+++ b/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/tools/TestStoragePolicyCommands.java
@@ -62,6 +62,69 @@ public class TestStoragePolicyCommands {
     }
   }
 
+
+  @Test
+  public void testSetAndUnsetStoragePolicy() throws Exception {
+    final Path foo = new Path("/foo");
+    final Path bar = new Path(foo, "bar");
+    final Path wow = new Path(bar, "wow");
+    DFSTestUtil.createFile(fs, wow, SIZE, REPL, 0);
+
+    /*
+     * test: set storage policy
+     */
+    final StoragePolicyAdmin admin = new StoragePolicyAdmin(conf);
+    DFSTestUtil.toolRun(admin, "-setStoragePolicy -path /foo -policy WARM", 0,
+        "Set storage policy WARM on " + foo.toString());
+    DFSTestUtil.toolRun(admin, "-setStoragePolicy -path /foo/bar -policy COLD",
+        0, "Set storage policy COLD on " + bar.toString());
+    DFSTestUtil.toolRun(admin, "-setStoragePolicy -path /foo/bar/wow -policy HOT",
+        0, "Set storage policy HOT on " + wow.toString());
+    DFSTestUtil.toolRun(admin, "-setStoragePolicy -path /fooz -policy WARM",
+        2, "File/Directory does not exist: /fooz");
+
+    /*
+     * test: get storage policy after set
+     */
+    final BlockStoragePolicySuite suite = BlockStoragePolicySuite
+        .createDefaultSuite();
+    final BlockStoragePolicy warm = suite.getPolicy("WARM");
+    final BlockStoragePolicy cold = suite.getPolicy("COLD");
+    final BlockStoragePolicy hot = suite.getPolicy("HOT");
+    DFSTestUtil.toolRun(admin, "-getStoragePolicy -path /foo", 0,
+        "The storage policy of " + foo.toString() + ":\n" + warm);
+    DFSTestUtil.toolRun(admin, "-getStoragePolicy -path /foo/bar", 0,
+        "The storage policy of " + bar.toString() + ":\n" + cold);
+    DFSTestUtil.toolRun(admin, "-getStoragePolicy -path /foo/bar/wow", 0,
+        "The storage policy of " + wow.toString() + ":\n" + hot);
+    DFSTestUtil.toolRun(admin, "-getStoragePolicy -path /fooz", 2,
+        "File/Directory does not exist: /fooz");
+
+    /*
+     * test: unset storage policy
+     */
+    DFSTestUtil.toolRun(admin, "-unsetStoragePolicy -path /foo", 0,
+        "Unset storage policy from " + foo.toString());
+    DFSTestUtil.toolRun(admin, "-unsetStoragePolicy -path /foo/bar", 0,
+        "Unset storage policy from " + bar.toString());
+    DFSTestUtil.toolRun(admin, "-unsetStoragePolicy -path /foo/bar/wow", 0,
+        "Unset storage policy from " + wow.toString());
+    DFSTestUtil.toolRun(admin, "-unsetStoragePolicy -path /fooz", 2,
+        "File/Directory does not exist: /fooz");
+
+    /*
+     * test: get storage policy after unset
+     */
+    DFSTestUtil.toolRun(admin, "-getStoragePolicy -path /foo", 0,
+        "The storage policy of " + foo.toString() + " is unspecified");
+    DFSTestUtil.toolRun(admin, "-getStoragePolicy -path /foo/bar", 0,
+        "The storage policy of " + bar.toString() + " is unspecified");
+    DFSTestUtil.toolRun(admin, "-getStoragePolicy -path /foo/bar/wow", 0,
+        "The storage policy of " + wow.toString() + " is unspecified");
+    DFSTestUtil.toolRun(admin, "-getStoragePolicy -path /fooz", 2,
+        "File/Directory does not exist: /fooz");
+  }
+
   @Test
   public void testSetAndGetStoragePolicy() throws Exception {
     final Path foo = new Path("/foo");