[
https://issues.apache.org/jira/browse/SPARK-3721?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
Josh Rosen resolved SPARK-3721.
-------------------------------
Resolution: Fixed
Fix Version/s: 1.2.0
Issue resolved by pull request 2659
[https://github.com/apache/spark/pull/2659]
> Broadcast Variables above 2GB break in PySpark
> ----------------------------------------------
>
> Key: SPARK-3721
> URL: https://issues.apache.org/jira/browse/SPARK-3721
> Project: Spark
> Issue Type: Bug
> Components: PySpark
> Affects Versions: 1.1.0
> Reporter: Brad Miller
> Assignee: Davies Liu
> Fix For: 1.2.0
>
>
> The bug displays 3 unique failure modes in PySpark, all of which seem to be
> related to broadcast variable size. Note that the tests below ran python
> 2.7.3 on all machines and used the Spark 1.1.0 binaries.
> **BLOCK 1** [no problem]
> {noformat}
> import cPickle
> from pyspark import SparkContext
> def check_pre_serialized(size):
> msg = cPickle.dumps(range(2 ** size))
> print 'serialized length:', len(msg)
> bvar = sc.broadcast(msg)
> print 'length recovered from broadcast variable:', len(bvar.value)
> print 'correct value recovered:', msg == bvar.value
> bvar.unpersist()
> def check_unserialized(size):
> msg = range(2 ** size)
> bvar = sc.broadcast(msg)
> print 'correct value recovered:', msg == bvar.value
> bvar.unpersist()
> SparkContext.setSystemProperty('spark.executor.memory', '15g')
> SparkContext.setSystemProperty('spark.cores.max', '5')
> sc = SparkContext('spark://crosby.research.intel-research.net:7077',
> 'broadcast_bug')
> {noformat}
> **BLOCK 2** [no problem]
> {noformat}
> check_pre_serialized(20)
> > serialized length: 9374656
> > length recovered from broadcast variable: 9374656
> > correct value recovered: True
> {noformat}
> **BLOCK 3** [no problem]
> {noformat}
> check_unserialized(20)
> > correct value recovered: True
> {noformat}
> **BLOCK 4** [no problem]
> {noformat}
> check_pre_serialized(27)
> > serialized length: 1499501632
> > length recovered from broadcast variable: 1499501632
> > correct value recovered: True
> {noformat}
> **BLOCK 5** [no problem]
> {noformat}
> check_unserialized(27)
> > correct value recovered: True
> {noformat}
> **BLOCK 6** **[ERROR 1: unhandled error from cPickle.dumps inside
> sc.broadcast]**
> {noformat}
> check_pre_serialized(28)
> .....
> > /home/spark/greatest/python/pyspark/serializers.py in dumps(self, obj)
> > 354
> > 355 def dumps(self, obj):
> > --> 356 return cPickle.dumps(obj, 2)
> > 357
> > 358 loads = cPickle.loads
> >
> > SystemError: error return without exception set
> {noformat}
> **BLOCK 7** [no problem]
> {noformat}
> check_unserialized(28)
> > correct value recovered: True
> {noformat}
> **BLOCK 8** **[ERROR 2: no error occurs and *incorrect result* is returned]**
> {noformat}
> check_pre_serialized(29)
> > serialized length: 6331339840
> > length recovered from broadcast variable: 2036372544
> > correct value recovered: False
> {noformat}
> **BLOCK 9** **[ERROR 3: unhandled error from zlib.compress inside
> sc.broadcast]**
> {noformat}
> check_unserialized(29)
> ......
> > /home/spark/greatest/python/pyspark/serializers.py in dumps(self, obj)
> > 418
> > 419 def dumps(self, obj):
> > --> 420 return zlib.compress(self.serializer.dumps(obj), 1)
> > 421
> > 422 def loads(self, obj):
> >
> > OverflowError: size does not fit in an int
> {noformat}
> **BLOCK 10** [ERROR 1]
> {noformat}
> check_pre_serialized(30)
> ...same as above...
> {noformat}
> **BLOCK 11** [ERROR 3]
> {noformat}
> check_unserialized(30)
> ...same as above...
> {noformat}
--
This message was sent by Atlassian JIRA
(v6.3.4#6332)
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscr...@spark.apache.org
For additional commands, e-mail: issues-h...@spark.apache.org
