From a7b3976bd4ce7f72a8476ccf5f9a8a06ff33bf21 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E2=80=9Cattilapiros=E2=80=9D?= Date: Thu, 12 Dec 2019 17:36:29 +0100 Subject: [PATCH 1/4] Initial upload. --- .../src/main/scala/org/apache/spark/storage/BlockManager.scala | 3 ++- docs/sql-migration-guide.md | 2 +- 2 files changed, 3 insertions(+), 2 deletions(-) diff --git a/core/src/main/scala/org/apache/spark/storage/BlockManager.scala b/core/src/main/scala/org/apache/spark/storage/BlockManager.scala index cc28f9b77da3..9cf513b7d94b 100644 --- a/core/src/main/scala/org/apache/spark/storage/BlockManager.scala +++ b/core/src/main/scala/org/apache/spark/storage/BlockManager.scala @@ -478,7 +478,8 @@ private[spark] class BlockManager( } hostLocalDirManager = - if (conf.get(config.SHUFFLE_HOST_LOCAL_DISK_READING_ENABLED)) { + if (conf.get(config.SHUFFLE_HOST_LOCAL_DISK_READING_ENABLED) && + !conf.get(config.SHUFFLE_USE_OLD_FETCH_PROTOCOL)) { externalBlockStoreClient.map { blockStoreClient => new HostLocalDirManager( futureExecutionContext, diff --git a/docs/sql-migration-guide.md b/docs/sql-migration-guide.md index ca78f3ca46ea..e34411e148bd 100644 --- a/docs/sql-migration-guide.md +++ b/docs/sql-migration-guide.md @@ -97,7 +97,7 @@ license: | - Since Spark 3.0, when Avro files are written with user provided non-nullable schema, even the catalyst schema is nullable, Spark is still able to write the files. However, Spark will throw runtime NPE if any of the records contains null. - - Since Spark 3.0, we use a new protocol for fetching shuffle blocks, for external shuffle service users, we need to upgrade the server correspondingly. Otherwise, we'll get the error message `UnsupportedOperationException: Unexpected message: FetchShuffleBlocks`. If it is hard to upgrade the shuffle service right now, you can still use the old protocol by setting `spark.shuffle.useOldFetchProtocol` to `true`. + - Since Spark 3.0, we use a new protocol for fetching shuffle blocks, for external shuffle service users, we need to upgrade the server correspondingly. Otherwise, we'll get the error message `IllegalArgumentException: Unexpected message type: `. If it is hard to upgrade the shuffle service right now, you can still use the old protocol by setting `spark.shuffle.useOldFetchProtocol` to `true`. - Since Spark 3.0, a higher-order function `exists` follows the three-valued boolean logic, i.e., if the `predicate` returns any `null`s and no `true` is obtained, then `exists` will return `null` instead of `false`. For example, `exists(array(1, null, 3), x -> x % 2 == 0)` will be `null`. The previous behaviour can be restored by setting `spark.sql.legacy.arrayExistsFollowsThreeValuedLogic` to `false`. From 7ca4c6d5d0d96743c5944a43cc21e42d652743ab Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E2=80=9Cattilapiros=E2=80=9D?= Date: Sun, 15 Dec 2019 12:03:27 +0100 Subject: [PATCH 2/4] extending doc of readHostLocalDisk --- .../scala/org/apache/spark/internal/config/package.scala | 6 +++--- 1 file changed, 3 insertions(+), 3 deletions(-) diff --git a/core/src/main/scala/org/apache/spark/internal/config/package.scala b/core/src/main/scala/org/apache/spark/internal/config/package.scala index 9d7b31aa30f0..e344a42b4f9a 100644 --- a/core/src/main/scala/org/apache/spark/internal/config/package.scala +++ b/core/src/main/scala/org/apache/spark/internal/config/package.scala @@ -1097,9 +1097,9 @@ package object config { private[spark] val SHUFFLE_HOST_LOCAL_DISK_READING_ENABLED = ConfigBuilder("spark.shuffle.readHostLocalDisk.enabled") - .doc("If enabled, shuffle blocks requested from those block managers which are running on " + - "the same host are read from the disk directly instead of being fetched as remote blocks " + - "over the network.") + .doc("If enabled (and `spark.shuffle.useOldFetchProtocol` is disabled), shuffle blocks " + + "requested from those block managers which are running on the same host are read from " + + "the disk directly instead of being fetched as remote blocks over the network.") .booleanConf .createWithDefault(true) From 69459e73457650ae43e08ef783b67cdf31b0d46f Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E2=80=9Cattilapiros=E2=80=9D?= Date: Tue, 17 Dec 2019 05:13:42 +0100 Subject: [PATCH 3/4] Applying review comments --- .../scala/org/apache/spark/internal/config/package.scala | 6 +++--- .../main/scala/org/apache/spark/storage/BlockManager.scala | 2 +- docs/core-migration-guide.md | 2 ++ docs/sql-migration-guide.md | 2 -- 4 files changed, 6 insertions(+), 6 deletions(-) diff --git a/core/src/main/scala/org/apache/spark/internal/config/package.scala b/core/src/main/scala/org/apache/spark/internal/config/package.scala index e344a42b4f9a..eba8199eacb2 100644 --- a/core/src/main/scala/org/apache/spark/internal/config/package.scala +++ b/core/src/main/scala/org/apache/spark/internal/config/package.scala @@ -1097,9 +1097,9 @@ package object config { private[spark] val SHUFFLE_HOST_LOCAL_DISK_READING_ENABLED = ConfigBuilder("spark.shuffle.readHostLocalDisk.enabled") - .doc("If enabled (and `spark.shuffle.useOldFetchProtocol` is disabled), shuffle blocks " + - "requested from those block managers which are running on the same host are read from " + - "the disk directly instead of being fetched as remote blocks over the network.") + .doc(s"If enabled (and `${SHUFFLE_USE_OLD_FETCH_PROTOCOL.key}` is disabled), shuffle " + + "blocks requested from those block managers which are running on the same host are read " + + "from the disk directly instead of being fetched as remote blocks over the network.") .booleanConf .createWithDefault(true) diff --git a/core/src/main/scala/org/apache/spark/storage/BlockManager.scala b/core/src/main/scala/org/apache/spark/storage/BlockManager.scala index 9cf513b7d94b..c47901314f53 100644 --- a/core/src/main/scala/org/apache/spark/storage/BlockManager.scala +++ b/core/src/main/scala/org/apache/spark/storage/BlockManager.scala @@ -479,7 +479,7 @@ private[spark] class BlockManager( hostLocalDirManager = if (conf.get(config.SHUFFLE_HOST_LOCAL_DISK_READING_ENABLED) && - !conf.get(config.SHUFFLE_USE_OLD_FETCH_PROTOCOL)) { + !conf.get(config.SHUFFLE_USE_OLD_FETCH_PROTOCOL)) { externalBlockStoreClient.map { blockStoreClient => new HostLocalDirManager( futureExecutionContext, diff --git a/docs/core-migration-guide.md b/docs/core-migration-guide.md index 17d071d0779b..fdb0afad6af9 100644 --- a/docs/core-migration-guide.md +++ b/docs/core-migration-guide.md @@ -36,3 +36,5 @@ license: | - Deprecated method `AccumulableInfo.apply` have been removed because creating `AccumulableInfo` is disallowed. - Event log file will be written as UTF-8 encoding, and Spark History Server will replay event log files as UTF-8 encoding. Previously Spark writes event log file as default charset of driver JVM process, so Spark History Server of Spark 2.x is needed to read the old event log files in case of incompatible encoding. + +- A new protocol for fetching shuffle blocks is used. It's recommended that external shuffle services be upgraded when running Spark 3.0 apps. Old external shuffle services can still be used by setting the configuration `spark.shuffle.useOldFetchProtocol` to `true`. Otherwise, Spark may run into errors with messages like `IllegalArgumentException: Unexpected message type: `. diff --git a/docs/sql-migration-guide.md b/docs/sql-migration-guide.md index e34411e148bd..73a4d19a1b15 100644 --- a/docs/sql-migration-guide.md +++ b/docs/sql-migration-guide.md @@ -97,8 +97,6 @@ license: | - Since Spark 3.0, when Avro files are written with user provided non-nullable schema, even the catalyst schema is nullable, Spark is still able to write the files. However, Spark will throw runtime NPE if any of the records contains null. - - Since Spark 3.0, we use a new protocol for fetching shuffle blocks, for external shuffle service users, we need to upgrade the server correspondingly. Otherwise, we'll get the error message `IllegalArgumentException: Unexpected message type: `. If it is hard to upgrade the shuffle service right now, you can still use the old protocol by setting `spark.shuffle.useOldFetchProtocol` to `true`. - - Since Spark 3.0, a higher-order function `exists` follows the three-valued boolean logic, i.e., if the `predicate` returns any `null`s and no `true` is obtained, then `exists` will return `null` instead of `false`. For example, `exists(array(1, null, 3), x -> x % 2 == 0)` will be `null`. The previous behaviour can be restored by setting `spark.sql.legacy.arrayExistsFollowsThreeValuedLogic` to `false`. - Since Spark 3.0, if files or subdirectories disappear during recursive directory listing (i.e. they appear in an intermediate listing but then cannot be read or listed during later phases of the recursive directory listing, due to either concurrent file deletions or object store consistency issues) then the listing will fail with an exception unless `spark.sql.files.ignoreMissingFiles` is `true` (default `false`). In previous versions, these missing files or subdirectories would be ignored. Note that this change of behavior only applies during initial table file listing (or during `REFRESH TABLE`), not during query execution: the net change is that `spark.sql.files.ignoreMissingFiles` is now obeyed during table file listing / query planning, not only at query execution time. From 6258ccd153fec744968b6a65f6f890dbfe013ad2 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E2=80=9Cattilapiros=E2=80=9D?= Date: Tue, 17 Dec 2019 07:46:09 +0100 Subject: [PATCH 4/4] move SHUFFLE_HOST_LOCAL_DISK_READING_ENABLED after SHUFFLE_USE_OLD_FETCH_PROTOCOL --- .../apache/spark/internal/config/package.scala | 16 ++++++++-------- 1 file changed, 8 insertions(+), 8 deletions(-) diff --git a/core/src/main/scala/org/apache/spark/internal/config/package.scala b/core/src/main/scala/org/apache/spark/internal/config/package.scala index eba8199eacb2..2078965e8824 100644 --- a/core/src/main/scala/org/apache/spark/internal/config/package.scala +++ b/core/src/main/scala/org/apache/spark/internal/config/package.scala @@ -1095,14 +1095,6 @@ package object config { .booleanConf .createWithDefault(false) - private[spark] val SHUFFLE_HOST_LOCAL_DISK_READING_ENABLED = - ConfigBuilder("spark.shuffle.readHostLocalDisk.enabled") - .doc(s"If enabled (and `${SHUFFLE_USE_OLD_FETCH_PROTOCOL.key}` is disabled), shuffle " + - "blocks requested from those block managers which are running on the same host are read " + - "from the disk directly instead of being fetched as remote blocks over the network.") - .booleanConf - .createWithDefault(true) - private[spark] val STORAGE_LOCAL_DISK_BY_EXECUTORS_CACHE_SIZE = ConfigBuilder("spark.storage.localDiskByExecutors.cacheSize") .doc("The max number of executors for which the local dirs are stored. This size is " + @@ -1148,6 +1140,14 @@ package object config { .booleanConf .createWithDefault(false) + private[spark] val SHUFFLE_HOST_LOCAL_DISK_READING_ENABLED = + ConfigBuilder("spark.shuffle.readHostLocalDisk.enabled") + .doc(s"If enabled (and `${SHUFFLE_USE_OLD_FETCH_PROTOCOL.key}` is disabled), shuffle " + + "blocks requested from those block managers which are running on the same host are read " + + "from the disk directly instead of being fetched as remote blocks over the network.") + .booleanConf + .createWithDefault(true) + private[spark] val MEMORY_MAP_LIMIT_FOR_TESTS = ConfigBuilder("spark.storage.memoryMapLimitForTests") .internal()