diff --git a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-nodemanager/src/main/java/org/apache/hadoop/yarn/server/nodemanager/containermanager/linux/runtime/DockerLinuxContainerRuntime.java b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-nodemanager/src/main/java/org/apache/hadoop/yarn/server/nodemanager/containermanager/linux/runtime/DockerLinuxContainerRuntime.java
index 681cae2cd90..43d8b4e4ac3 100644
--- a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-nodemanager/src/main/java/org/apache/hadoop/yarn/server/nodemanager/containermanager/linux/runtime/DockerLinuxContainerRuntime.java
+++ b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-nodemanager/src/main/java/org/apache/hadoop/yarn/server/nodemanager/containermanager/linux/runtime/DockerLinuxContainerRuntime.java
@@ -45,11 +45,14 @@ import org.apache.hadoop.yarn.server.nodemanager.containermanager.runtime.Contai
 import org.apache.hadoop.yarn.server.nodemanager.containermanager.runtime.ContainerRuntimeContext;
 
 
+import java.nio.file.Files;
+import java.nio.file.Paths;
 import java.util.ArrayList;
 import java.util.Arrays;
 import java.util.HashSet;
 import java.util.List;
 import java.util.Map;
+import java.util.Map.Entry;
 import java.util.Set;
 
 import static org.apache.hadoop.yarn.server.nodemanager.containermanager.linux.runtime.LinuxContainerRuntimeConstants.*;
@@ -72,6 +75,9 @@ public class DockerLinuxContainerRuntime implements LinuxContainerRuntime {
   @InterfaceAudience.Private
   public static final String ENV_DOCKER_CONTAINER_RUN_PRIVILEGED_CONTAINER =
       "YARN_CONTAINER_RUNTIME_DOCKER_RUN_PRIVILEGED_CONTAINER";
+  @InterfaceAudience.Private
+  public static final String ENV_DOCKER_CONTAINER_LOCAL_RESOURCE_MOUNTS =
+      "YARN_CONTAINER_RUNTIME_DOCKER_LOCAL_RESOURCE_MOUNTS";
 
   private Configuration conf;
   private DockerClient dockerClient;
@@ -225,6 +231,27 @@ public class DockerLinuxContainerRuntime implements LinuxContainerRuntime {
     return true;
   }
 
+  @VisibleForTesting
+  protected String validateMount(String mount,
+      Map<Path, List<String>> localizedResources)
+      throws ContainerExecutionException {
+    for (Entry<Path, List<String>> resource : localizedResources.entrySet()) {
+      if (resource.getValue().contains(mount)) {
+        java.nio.file.Path path = Paths.get(resource.getKey().toString());
+        if (!path.isAbsolute()) {
+          throw new ContainerExecutionException("Mount must be absolute: " +
+              mount);
+        }
+        if (Files.isSymbolicLink(path)) {
+          throw new ContainerExecutionException("Mount cannot be a symlink: " +
+              mount);
+        }
+        return path.toString();
+      }
+    }
+    throw new ContainerExecutionException("Mount must be a localized " +
+        "resource: " + mount);
+  }
 
   @Override
   public void launchContainer(ContainerRuntimeContext ctx)
@@ -254,6 +281,9 @@ public class DockerLinuxContainerRuntime implements LinuxContainerRuntime {
     @SuppressWarnings("unchecked")
     List<String> containerLogDirs = ctx.getExecutionAttribute(
         CONTAINER_LOG_DIRS);
+    @SuppressWarnings("unchecked")
+    Map<Path, List<String>> localizedResources = ctx.getExecutionAttribute(
+        LOCALIZED_RESOURCES);
     Set<String> capabilities = new HashSet<>(Arrays.asList(conf.getStrings(
         YarnConfiguration.NM_DOCKER_CONTAINER_CAPABILITIES,
         YarnConfiguration.DEFAULT_NM_DOCKER_CONTAINER_CAPABILITIES)));
@@ -274,6 +304,23 @@ public class DockerLinuxContainerRuntime implements LinuxContainerRuntime {
       runCommand.addMountLocation(dir, dir);
     }
 
+    if (environment.containsKey(ENV_DOCKER_CONTAINER_LOCAL_RESOURCE_MOUNTS)) {
+      String mounts = environment.get(
+          ENV_DOCKER_CONTAINER_LOCAL_RESOURCE_MOUNTS);
+      if (!mounts.isEmpty()) {
+        for (String mount : StringUtils.split(mounts)) {
+          String[] dir = StringUtils.split(mount, ':');
+          if (dir.length != 2) {
+            throw new ContainerExecutionException("Invalid mount : " +
+                mount);
+          }
+          String src = validateMount(dir[0], localizedResources);
+          String dst = dir[1];
+          runCommand.addMountLocation(src, dst + ":ro");
+        }
+      }
+    }
+
     if (allowPrivilegedContainerExecution(container)) {
       runCommand.setPrivileged();
     }
diff --git a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-nodemanager/src/test/java/org/apache/hadoop/yarn/server/nodemanager/containermanager/linux/runtime/TestDockerContainerRuntime.java b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-nodemanager/src/test/java/org/apache/hadoop/yarn/server/nodemanager/containermanager/linux/runtime/TestDockerContainerRuntime.java
index d1bdabec306..538b03fcbb4 100644
--- a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-nodemanager/src/test/java/org/apache/hadoop/yarn/server/nodemanager/containermanager/linux/runtime/TestDockerContainerRuntime.java
+++ b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-nodemanager/src/test/java/org/apache/hadoop/yarn/server/nodemanager/containermanager/linux/runtime/TestDockerContainerRuntime.java
@@ -49,6 +49,7 @@ import java.nio.file.Files;
 import java.nio.file.Paths;
 import java.util.ArrayList;
 import java.util.Arrays;
+import java.util.Collections;
 import java.util.HashMap;
 import java.util.HashSet;
 import java.util.List;
@@ -83,6 +84,7 @@ public class TestDockerContainerRuntime {
   List<String> logDirs;
   List<String> containerLocalDirs;
   List<String> containerLogDirs;
+  Map<Path,List<String>> localizedResources;
   String resourcesOptions;
   ContainerRuntimeContext.Builder builder;
   String submittingUser = "anakin";
@@ -127,11 +129,14 @@ public class TestDockerContainerRuntime {
     resourcesOptions = "cgroups=none";
     containerLocalDirs = new ArrayList<>();
     containerLogDirs = new ArrayList<>();
+    localizedResources = new HashMap<>();
 
     localDirs.add("/test_local_dir");
     logDirs.add("/test_log_dir");
     containerLocalDirs.add("/test_container_local_dir");
     containerLogDirs.add("/test_container_log_dir");
+    localizedResources.put(new Path("/test_local_dir/test_resource_file"),
+        Collections.singletonList("test_dir/test_resource_file"));
 
     builder = new ContainerRuntimeContext
         .Builder(container);
@@ -149,6 +154,7 @@ public class TestDockerContainerRuntime {
         .setExecutionAttribute(LOG_DIRS, logDirs)
         .setExecutionAttribute(CONTAINER_LOCAL_DIRS, containerLocalDirs)
         .setExecutionAttribute(CONTAINER_LOG_DIRS, containerLogDirs)
+        .setExecutionAttribute(LOCALIZED_RESOURCES, localizedResources)
         .setExecutionAttribute(RESOURCES_OPTIONS, resourcesOptions);
   }
 
@@ -445,4 +451,113 @@ public class TestDockerContainerRuntime {
     //no --cgroup-parent should be added in either case
     Mockito.verifyZeroInteractions(command);
   }
+
+  @Test
+  public void testMountSourceOnly()
+      throws ContainerExecutionException, PrivilegedOperationException,
+      IOException{
+    DockerLinuxContainerRuntime runtime = new DockerLinuxContainerRuntime(
+        mockExecutor, mockCGroupsHandler);
+    runtime.initialize(conf);
+
+    env.put(
+        DockerLinuxContainerRuntime.ENV_DOCKER_CONTAINER_LOCAL_RESOURCE_MOUNTS,
+        "source");
+
+    try {
+      runtime.launchContainer(builder.build());
+      Assert.fail("Expected a launch container failure due to invalid mount.");
+    } catch (ContainerExecutionException e) {
+      LOG.info("Caught expected exception : " + e);
+    }
+  }
+
+  @Test
+  public void testMountSourceTarget()
+      throws ContainerExecutionException, PrivilegedOperationException,
+      IOException{
+    DockerLinuxContainerRuntime runtime = new DockerLinuxContainerRuntime(
+        mockExecutor, mockCGroupsHandler);
+    runtime.initialize(conf);
+
+    env.put(
+        DockerLinuxContainerRuntime.ENV_DOCKER_CONTAINER_LOCAL_RESOURCE_MOUNTS,
+        "test_dir/test_resource_file:test_mount");
+
+    runtime.launchContainer(builder.build());
+    PrivilegedOperation op = capturePrivilegedOperationAndVerifyArgs();
+    List<String> args = op.getArguments();
+    String dockerCommandFile = args.get(11);
+
+    List<String> dockerCommands = Files.readAllLines(Paths.get
+        (dockerCommandFile), Charset.forName("UTF-8"));
+
+    Assert.assertEquals(1, dockerCommands.size());
+
+    String command = dockerCommands.get(0);
+
+    Assert.assertTrue("Did not find expected " +
+        "/test_local_dir/test_resource_file:test_mount mount in docker " +
+        "run args : " + command,
+        command.contains(" -v /test_local_dir/test_resource_file:test_mount" +
+            ":ro "));
+  }
+
+  @Test
+  public void testMountInvalid()
+      throws ContainerExecutionException, PrivilegedOperationException,
+      IOException{
+    DockerLinuxContainerRuntime runtime = new DockerLinuxContainerRuntime(
+        mockExecutor, mockCGroupsHandler);
+    runtime.initialize(conf);
+
+    env.put(
+        DockerLinuxContainerRuntime.ENV_DOCKER_CONTAINER_LOCAL_RESOURCE_MOUNTS,
+        "source:target:other");
+
+    try {
+      runtime.launchContainer(builder.build());
+      Assert.fail("Expected a launch container failure due to invalid mount.");
+    } catch (ContainerExecutionException e) {
+      LOG.info("Caught expected exception : " + e);
+    }
+  }
+
+  @Test
+  public void testMountMultiple()
+      throws ContainerExecutionException, PrivilegedOperationException,
+      IOException{
+    DockerLinuxContainerRuntime runtime = new DockerLinuxContainerRuntime(
+        mockExecutor, mockCGroupsHandler);
+    runtime.initialize(conf);
+
+    env.put(
+        DockerLinuxContainerRuntime.ENV_DOCKER_CONTAINER_LOCAL_RESOURCE_MOUNTS,
+        "test_dir/test_resource_file:test_mount1," +
+            "test_dir/test_resource_file:test_mount2");
+
+    runtime.launchContainer(builder.build());
+    PrivilegedOperation op = capturePrivilegedOperationAndVerifyArgs();
+    List<String> args = op.getArguments();
+    String dockerCommandFile = args.get(11);
+
+    List<String> dockerCommands = Files.readAllLines(Paths.get
+        (dockerCommandFile), Charset.forName("UTF-8"));
+
+    Assert.assertEquals(1, dockerCommands.size());
+
+    String command = dockerCommands.get(0);
+
+    Assert.assertTrue("Did not find expected " +
+        "/test_local_dir/test_resource_file:test_mount1 mount in docker " +
+        "run args : " + command,
+        command.contains(" -v /test_local_dir/test_resource_file:test_mount1" +
+            ":ro "));
+    Assert.assertTrue("Did not find expected " +
+        "/test_local_dir/test_resource_file:test_mount2 mount in docker " +
+        "run args : " + command,
+        command.contains(" -v /test_local_dir/test_resource_file:test_mount2" +
+            ":ro "));
+  }
+
 }