Skip to content

[SPARK-30235][CORE] Switching off host local disk reading of shuffle blocks in case of useOldFetchProtocol #26869

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Closed
attilapiros wants to merge 4 commits into from
Closed

[SPARK-30235][CORE] Switching off host local disk reading of shuffle blocks in case of useOldFetchProtocol #26869

Show file tree
Hide file tree
Changes from 2 commits
Commits
Show all changes
4 commits
Select commit Hold shift + click to select a range
a7b3976
Initial upload.
attilapiros Dec 12, 2019
7ca4c6d
extending doc of readHostLocalDisk
attilapiros Dec 15, 2019
69459e7
Applying review comments
attilapiros Dec 17, 2019
6258ccd
move SHUFFLE_HOST_LOCAL_DISK_READING_ENABLED after SHUFFLE_USE_OLD_FE…
attilapiros Dec 17, 2019
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
6 changes: 3 additions & 3 deletions core/src/main/scala/org/apache/spark/internal/config/package.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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)

Expand Down
3 changes: 2 additions & 1 deletion core/src/main/scala/org/apache/spark/storage/BlockManager.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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,
Expand Down
2 changes: 1 addition & 1 deletion docs/sql-migration-guide.md
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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: <number>`. 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`.

Expand Down