You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@flink.apache.org by NicoK <gi...@git.apache.org> on 2017/12/18 13:43:35 UTC
[GitHub] flink pull request #5176: [FLINK-8279][blob] fall back to TaskManager temp d...
GitHub user NicoK opened a pull request:
https://github.com/apache/flink/pull/5176
[FLINK-8279][blob] fall back to TaskManager temp directories first
## What is the purpose of the change
Instead of falling back to `java.io.tmpdir` directly, the BLOB server and cache processes should fall back to the TaskManager temp directories (given by `ConfigConstants#TASK_MANAGER_TMP_DIR_KEY}` directly, before falling back
to `ConfigConstants#DEFAULT_TASK_MANAGER_TMP_PATH` (set to `java.io.tmpdir`).
In a Mesos/YARN environment, this means that we will use the designated temp directories for our jobs instead of the system-wide `java.io.tmpdir`. These directories may also offer some more space.
## Brief change log
- change `BlobUtils#initLocalStorageDirectory()` storage directory initialisation to use temporary directories in this order (and fall back to the next if not available):
- `BlobServerOptions.STORAGE_DIRECTORY`
- `ConfigConstants.TASK_MANAGER_TMP_DIR_KEY` (and randomly select one among them)
- `ConfigConstants.DEFAULT_TASK_MANAGER_TMP_PATH`
## Verifying this change
This change added tests and can be verified as follows:
- added tests verifying the fallbacks
## Does this pull request potentially affect one of the following parts:
- Dependencies (does it add or upgrade a dependency): **no**
- The public API, i.e., is any changed class annotated with `@Public(Evolving)`: **no**
- The serializers: **no**
- The runtime per-record code paths (performance sensitive): **no**
- Anything that affects deployment or recovery: JobManager (and its components), Checkpointing, Yarn/Mesos, ZooKeeper: **no**
- The S3 file system connector: **no**
## Documentation
- Does this pull request introduce a new feature? **no**
- If yes, how is the feature documented? **Docs, JavaDocs**
You can merge this pull request into a Git repository by running:
$ git pull https://github.com/NicoK/flink flink-8279
Alternatively you can review and apply these changes as the patch at:
https://github.com/apache/flink/pull/5176.patch
To close this pull request, make a commit to your master/trunk branch
with (at least) the following in the commit message:
This closes #5176
----
commit 08e539c9abf825aa1ca60574cd9e0c5724cee252
Author: Nico Kruber <ni...@data-artisans.com>
Date: 2017-12-18T13:19:02Z
[FLINK-8279][blob] fall back to TaskManager temp directories first
Instead of falling back to java.io.tmpdir directly, the BLOB server and cache
processes fall back to the TaskManager temp directories (given by
ConfigConstants#TASK_MANAGER_TMP_DIR_KEY) directly, before falling back
to ConfigConstants#DEFAULT_TASK_MANAGER_TMP_PATH (set to java.io.tmpdir).
In a Mesos/YARN environment, this means that we will use the designated temp
directories for our jobs instead of the system-wide java.io.tmpdir. These
directories may also offer some more space.
----
---
[GitHub] flink issue #5176: [FLINK-8279][blob] fall back to TaskManager temp director...
Posted by EronWright <gi...@git.apache.org>.
Github user EronWright commented on the issue:
https://github.com/apache/flink/pull/5176
@NicoK was this PR a response to an actual issue, and did it resolve it?
Do I understand correctly that this will cause the JM (blob server) to use `taskmanager.tmp.dirs`? In addition to the TM blob cache, of course.
For the TM on YARN, is there really any effect? Looking at MAPREDUCE-6472, seems the temp folder is already within the container. Can you explain the actual effect? Thanks.
Note that the Mesos implementation doesn't actually configure `taskmanager.tmp.dirs` at this time (though there's some dead code in TM). A proper treatment of tmp folders on Mesos would involve the use of volumes.
---
[GitHub] flink pull request #5176: [FLINK-8279][blob] fall back to TaskManager temp d...
Posted by StefanRRichter <gi...@git.apache.org>.
Github user StefanRRichter commented on a diff in the pull request:
https://github.com/apache/flink/pull/5176#discussion_r160155461
--- Diff: flink-runtime/src/main/java/org/apache/flink/runtime/blob/BlobUtils.java ---
@@ -127,21 +132,28 @@ private static BlobStoreService createFileSystemBlobStore(Configuration configur
}
/**
- * Creates a local storage directory for a blob service under the given parent directory.
+ * Creates a local storage directory for a blob service under the configuration parameter given
+ * by {@link BlobServerOptions#STORAGE_DIRECTORY}. If this is <tt>null</tt> or empty, we will
+ * fall back to Flink's temp directories (given by
+ * {@link org.apache.flink.configuration.CoreOptions#TMP_DIRS}) and choose one among them at
+ * random.
*
- * @param basePath
- * base path, i.e. parent directory, of the storage directory to use (if <tt>null</tt> or
- * empty, the path in <tt>java.io.tmpdir</tt> will be used)
+ * @param config
+ * Flink configuration
*
* @return a new local storage directory
*
* @throws IOException
* thrown if the local file storage cannot be created or is not usable
*/
- static File initLocalStorageDirectory(String basePath) throws IOException {
+ static File initLocalStorageDirectory(Configuration config) throws IOException {
+
+ String basePath = config.getString(BlobServerOptions.STORAGE_DIRECTORY);
+
File baseDir;
if (StringUtils.isNullOrWhitespaceOnly(basePath)) {
- baseDir = new File(System.getProperty("java.io.tmpdir"));
+ final String[] tmpDirPaths = TaskManagerServicesConfiguration.parseTempDirectories(config);
+ baseDir = new File(tmpDirPaths[rnd.nextInt(tmpDirPaths.length)]);
--- End diff --
I wonder if this is the best strategy and if we should only return a single directory, which might represent only a fraction of the available space. However, I would assume in most cases user only specify a single directory anyways. So what to do about this, is it feasible that the blob cache can deal with all the offered directories?
---
[GitHub] flink issue #5176: [FLINK-8279][blob] fall back to TaskManager temp director...
Posted by tillrohrmann <gi...@git.apache.org>.
Github user tillrohrmann commented on the issue:
https://github.com/apache/flink/pull/5176
There are checkstyle violations. Could you please fix them @NicoK.
---
[GitHub] flink pull request #5176: [FLINK-8279][blob] fall back to TaskManager temp d...
Posted by asfgit <gi...@git.apache.org>.
Github user asfgit closed the pull request at:
https://github.com/apache/flink/pull/5176
---
[GitHub] flink issue #5176: [FLINK-8279][blob] fall back to TaskManager temp director...
Posted by NicoK <gi...@git.apache.org>.
Github user NicoK commented on the issue:
https://github.com/apache/flink/pull/5176
For the change regarding the JobManager use of a TaskManager configuration parameter, I created a separate issue and included the appropriate commits here (FLINK-8250) as this commit probably does not make too much sense without them. The commits contain some high-level description about the changes.
---
[GitHub] flink pull request #5176: [FLINK-8279][blob] fall back to TaskManager temp d...
Posted by NicoK <gi...@git.apache.org>.
Github user NicoK commented on a diff in the pull request:
https://github.com/apache/flink/pull/5176#discussion_r160353839
--- Diff: flink-runtime/src/main/java/org/apache/flink/runtime/blob/BlobUtils.java ---
@@ -127,21 +132,28 @@ private static BlobStoreService createFileSystemBlobStore(Configuration configur
}
/**
- * Creates a local storage directory for a blob service under the given parent directory.
+ * Creates a local storage directory for a blob service under the configuration parameter given
+ * by {@link BlobServerOptions#STORAGE_DIRECTORY}. If this is <tt>null</tt> or empty, we will
+ * fall back to Flink's temp directories (given by
+ * {@link org.apache.flink.configuration.CoreOptions#TMP_DIRS}) and choose one among them at
+ * random.
*
- * @param basePath
- * base path, i.e. parent directory, of the storage directory to use (if <tt>null</tt> or
- * empty, the path in <tt>java.io.tmpdir</tt> will be used)
+ * @param config
+ * Flink configuration
*
* @return a new local storage directory
*
* @throws IOException
* thrown if the local file storage cannot be created or is not usable
*/
- static File initLocalStorageDirectory(String basePath) throws IOException {
+ static File initLocalStorageDirectory(Configuration config) throws IOException {
+
+ String basePath = config.getString(BlobServerOptions.STORAGE_DIRECTORY);
+
File baseDir;
if (StringUtils.isNullOrWhitespaceOnly(basePath)) {
- baseDir = new File(System.getProperty("java.io.tmpdir"));
+ final String[] tmpDirPaths = TaskManagerServicesConfiguration.parseTempDirectories(config);
+ baseDir = new File(tmpDirPaths[rnd.nextInt(tmpDirPaths.length)]);
--- End diff --
Unfortunately, the BLOB caches cannot handle multiple directories that well. When looking for cached files, they check for their existence and otherwise download missing ones. Looking through all of them may be cumbersome as well (and should not be solved with this PR, imho).
I'd also expect most use cases to use a single directory but tbh, I lack overview of use cases here.
---
[GitHub] flink issue #5176: [FLINK-8279][blob] fall back to TaskManager temp director...
Posted by NicoK <gi...@git.apache.org>.
Github user NicoK commented on the issue:
https://github.com/apache/flink/pull/5176
@EronWright this PR originated from a user problem at https://lists.apache.org/thread.html/177644ad38af34fb7ae7fb12a3bfcef3a70cd70dfccb6f7008a24eb6@%3Cuser.flink.apache.org%3E
And yes, this also adds the fallback to (a subdirectory in) `taskmanager.tmp.dirs` in case no `blob.storage.directory` was given to the JobManager. Given that the configuration parameter hints towards the TaskManager only, we should maybe rename it?
Regarding [MAPREDUCE-6472](https://issues.apache.org/jira/browse/MAPREDUCE-6472), it seems this is only available from Hadoop 2.7.2 onwards. With this fix applied, the behaviour changes only slightly: instead of using `java.io.tmpdir` directly (set to `Environment.PWD/tmp`), this will use the `taskmanager.tmp.dirs` which is set to `Environment.LOCAL_DIRS` by `YarnTaskManagerRunner`/`YarnTaskExecutorRunner`.
I adapted the appropriate code paths for Mesos as well. If this is not used yet, it will at least work in the future once the `taskmanager.tmp.dirs` handling is correct. Until then, the fallback of this to `java.io.tmpdir` seems to be working as well (if this is set appropriately by Mesos) - but this is not only a problem of the BLOB server parts then and is out of the scope of this PR.
---
[GitHub] flink pull request #5176: [FLINK-8279][blob] fall back to TaskManager temp d...
Posted by EronWright <gi...@git.apache.org>.
Github user EronWright commented on a diff in the pull request:
https://github.com/apache/flink/pull/5176#discussion_r157548338
--- Diff: flink-runtime/src/main/java/org/apache/flink/runtime/blob/BlobUtils.java ---
@@ -127,21 +132,31 @@ private static BlobStoreService createFileSystemBlobStore(Configuration configur
}
/**
- * Creates a local storage directory for a blob service under the given parent directory.
+ * Creates a local storage directory for a blob service under the configuration parameter given
+ * by {@link BlobServerOptions#STORAGE_DIRECTORY}. If this is <tt>null</tt> or empty, we will
+ * fall back to the TaskManager temp directories (given by
+ * {@link ConfigConstants#TASK_MANAGER_TMP_DIR_KEY}; which in turn falls back to
+ * {@link ConfigConstants#DEFAULT_TASK_MANAGER_TMP_PATH} currently set to
+ * <tt>java.io.tmpdir</tt>) and choose one among them at random.
*
- * @param basePath
- * base path, i.e. parent directory, of the storage directory to use (if <tt>null</tt> or
- * empty, the path in <tt>java.io.tmpdir</tt> will be used)
+ * @param config
+ * Flink configuration
*
* @return a new local storage directory
*
* @throws IOException
* thrown if the local file storage cannot be created or is not usable
*/
- static File initLocalStorageDirectory(String basePath) throws IOException {
+ static File initLocalStorageDirectory(Configuration config) throws IOException {
+
+ String basePath = config.getString(BlobServerOptions.STORAGE_DIRECTORY);
+
File baseDir;
if (StringUtils.isNullOrWhitespaceOnly(basePath)) {
- baseDir = new File(System.getProperty("java.io.tmpdir"));
+ final String[] tmpDirPaths = config.getString(
+ ConfigConstants.TASK_MANAGER_TMP_DIR_KEY,
+ ConfigConstants.DEFAULT_TASK_MANAGER_TMP_PATH).split(",|" + File.pathSeparator);
--- End diff --
Consider encapsulating this parsing logic into `TaskManagerServicesConfiguration` or similar.
---
[GitHub] flink pull request #5176: [FLINK-8279][blob] fall back to TaskManager temp d...
Posted by StefanRRichter <gi...@git.apache.org>.
Github user StefanRRichter commented on a diff in the pull request:
https://github.com/apache/flink/pull/5176#discussion_r160150281
--- Diff: flink-runtime/src/main/java/org/apache/flink/runtime/blob/BlobUtils.java ---
@@ -72,6 +75,8 @@
*/
static final String NO_JOB_DIR_PREFIX = "no_job";
+ private static final Random rnd = new Random();
--- End diff --
By convention, this field name should be all uppercase.
---