Skip to content

[SPARK-28355][CORE][PYTHON] Use Spark conf for threshold at which command is compressed by broadcast #25123

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
jessecai wants to merge 5 commits into from
Closed

[SPARK-28355][CORE][PYTHON] Use Spark conf for threshold at which command is compressed by broadcast #25123

Show file tree
Hide file tree
Changes from 2 commits
Commits
Show all changes
5 commits
Select commit Hold shift + click to select a range
2590b59
Add UDF threshold conf and replace in _prepare_for_python_RDD
jessecai Jul 11, 2019
ca7de40
Address comments
jessecai Jul 12, 2019
c716eb7
Fix description
jessecai Jul 12, 2019
f34d3bd
Add test case
jessecai Jul 12, 2019
1ec3a4a
Add test for the default value
jessecai Jul 12, 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
4 changes: 4 additions & 0 deletions core/src/main/scala/org/apache/spark/api/python/PythonUtils.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -81,4 +81,8 @@ private[spark] object PythonUtils {
def isEncryptionEnabled(sc: JavaSparkContext): Boolean = {
sc.conf.get(org.apache.spark.internal.config.IO_ENCRYPTION_ENABLED)
}

def getBroadcastThreshold(sc: JavaSparkContext): Long = {
sc.conf.get(org.apache.spark.internal.config.BROADCAST_FOR_UDF_COMPRESSION_THRESHOLD)
}
}
8 changes: 8 additions & 0 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 @@ -1246,6 +1246,14 @@ package object config {
"mechanisms to guarantee data won't be corrupted during broadcast")
.booleanConf.createWithDefault(true)

private[spark] val BROADCAST_FOR_UDF_COMPRESSION_THRESHOLD =
ConfigBuilder("spark.broadcast.UDFCompressionThreshold")
.doc("The threshold at which a a user-defined function (UDF) is compressed by broadcast, " +
Copy link
Member

@HyukjinKwon HyukjinKwon Jul 12, 2019

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I think this also applies to RDD APIs. We can just say, for instance, The threshold at which Python commands for RDD APIs and user-defined function (UDF) are serialized by broadcast .... Feel free to change wording.

Copy link
Contributor Author

@jessecai jessecai Jul 12, 2019

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I updated the description to include this.

"in bytes unless otherwise specified")
.bytesConf(ByteUnit.BYTE)
.checkValue(v => v >= 0, "The threshold should be non-negative.")
.createWithDefault(1L * 1024 * 1024)

private[spark] val RDD_COMPRESS = ConfigBuilder("spark.rdd.compress")
.doc("Whether to compress serialized RDD partitions " +
"(e.g. for StorageLevel.MEMORY_ONLY_SER in Scala " +
Expand Down
2 changes: 1 addition & 1 deletion python/pyspark/rdd.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -2490,7 +2490,7 @@ def _prepare_for_python_RDD(sc, command):
# the serialized command will be compressed by broadcast
ser = CloudPickleSerializer()
pickled_command = ser.dumps(command)
if len(pickled_command) > (1 << 20): # 1M
if len(pickled_command) > sc._jvm.PythonUtils.getBroadcastThreshold(sc._jsc): # Default 1M
# The broadcast will have same life cycle as created PythonRDD
broadcast = sc.broadcast(pickled_command)
pickled_command = ser.dumps(broadcast)
Expand Down