{"push_id":730586415,"size":1,"distinct_size":1,"ref":"refs/heads/master","head":"a803ac3e060d181c7b34d9501c9350e5f215ba85","before":"34ed82bb44c4519819695ddc760e6c9a98bc2e40","commits":[{"sha":"a803ac3e060d181c7b34d9501c9350e5f215ba85","author":{"email":"d14f037939c13b3e76faad4b1a2388cbb2de1ccb@gmail.com","name":"Nicholas Hwang"},"message":"[SPARK-9021] [PYSPARK] Change RDD.aggregate() to do reduce(mapPartitions()) instead of mapPartitions.fold()\n\nI'm relatively new to Spark and functional programming, so forgive me if this pull request is just a result of my misunderstanding of how Spark should be used.\n\nCurrently, if one happens to use a mutable object as `zeroValue` for `RDD.aggregate()`, possibly unexpected behavior can occur.\n\nThis is because pyspark's current implementation of `RDD.aggregate()` does not serialize or make a copy of `zeroValue` before handing it off to `RDD.mapPartitions(...).fold(...)`. This results in a single reference to `zeroValue` being used for both `RDD.mapPartitions()` and `RDD.fold()` on each partition. This can result in strange accumulator values being fed into each partition's call to `RDD.fold()`, as the `zeroValue` may have been changed in-place during the `RDD.mapPartitions()` call.\n\nAs an illustrative example, submit the following to `spark-submit`:\n```\nfrom pyspark import SparkConf, SparkContext\nimport collections\n\ndef updateCounter(acc, val):\n    print 'update acc:', acc\n    print 'update val:', val\n    acc[val] += 1\n    return acc\n\ndef comboCounter(acc1, acc2):\n    print 'combo acc1:', acc1\n    print 'combo acc2:', acc2\n    acc1.update(acc2)\n    return acc1\n\ndef main():\n    conf = SparkConf().setMaster(\"local\").setAppName(\"Aggregate with Counter\")\n    sc = SparkContext(conf = conf)\n\n    print '======= AGGREGATING with ONE PARTITION ======='\n    print sc.parallelize(range(1,10), 1).aggregate(collections.Counter(), updateCounter, comboCounter)\n\n    print '======= AGGREGATING with TWO PARTITIONS ======='\n    print sc.parallelize(range(1,10), 2).aggregate(collections.Counter(), updateCounter, comboCounter)\n\nif __name__ == \"__main__\":\n    main()\n```\n\nOne probably expects this to output the following:\n```\nCounter({1: 1, 2: 1, 3: 1, 4: 1, 5: 1, 6: 1, 7: 1, 8: 1, 9: 1})\n```\n\nBut it instead outputs this (regardless of the number of partitions):\n```\nCounter({1: 2, 2: 2, 3: 2, 4: 2, 5: 2, 6: 2, 7: 2, 8: 2, 9: 2})\n```\n\nThis is because (I believe) `zeroValue` gets passed correctly to each partition, but after `RDD.mapPartitions()` completes, the `zeroValue` object has been updated and is then passed to `RDD.fold()`, which results in all items being double-counted within each partition before being finally reduced at the calling node.\n\nI realize that this type of calculation is typically done by `RDD.mapPartitions(...).reduceByKey(...)`, but hopefully this illustrates some potentially confusing behavior. I also noticed that other `RDD` methods use this `deepcopy` approach to creating unique copies of `zeroValue` (i.e., `RDD.aggregateByKey()` and `RDD.foldByKey()`), and that the Scala implementations do seem to serialize the `zeroValue` object appropriately to prevent this type of behavior.\n\nAuthor: Nicholas Hwang <moogling@gmail.com>\n\nCloses #7378 from njhwang/master and squashes the following commits:\n\n659bb27 [Nicholas Hwang] Fixed RDD.aggregate() to perform a reduce operation on collected mapPartitions results, similar to how fold currently is implemented. This prevents an initial combOp being performed on each partition with zeroValue (which leads to unexpected behavior if zeroValue is a mutable object) before being combOp'ed with other partition results.\n8d8d694 [Nicholas Hwang] Changed dict construction to be compatible with Python 2.6 (cannot use list comprehensions to make dicts)\n56eb2ab [Nicholas Hwang] Fixed whitespace after colon to conform with PEP8\n391de4a [Nicholas Hwang] Removed used of collections.Counter from RDD tests for Python 2.6 compatibility; used defaultdict(int) instead. Merged treeAggregate test with mutable zero value into aggregate test to reduce code duplication.\n2fa4e4b [Nicholas Hwang] Merge branch 'master' of https://github.com/njhwang/spark\nba528bd [Nicholas Hwang] Updated comments regarding protection of zeroValue from mutation in RDD.aggregate(). Added regression tests for aggregate(), fold(), aggregateByKey(), foldByKey(), and treeAggregate(), all with both 1 and 2 partition RDDs. Confirmed that aggregate() is the only problematic implementation as of commit 257236c3e17906098f801cbc2059e7a9054e8cab. Also replaced some parallelizations of ranges with xranges, per the documentation's recommendations of preferring xrange over range.\n7820391 [Nicholas Hwang] Updated comments regarding protection of zeroValue from mutation in RDD.aggregate(). Added regression tests for aggregate(), fold(), aggregateByKey(), foldByKey(), and treeAggregate(), all with both 1 and 2 partition RDDs. Confirmed that aggregate() is the only problematic implementation as of commit 257236c3e17906098f801cbc2059e7a9054e8cab.\n90d1544 [Nicholas Hwang] Made sure RDD.aggregate() makes a deepcopy of zeroValue for all partitions; this ensures that the mapPartitions call works with unique copies of zeroValue in each partition, and prevents a single reference to zeroValue being used for both map and fold calls on each partition (resulting in possibly unexpected behavior).","distinct":true,"url":"https://api.github.com/repos/ueshin/apache-spark/commits/a803ac3e060d181c7b34d9501c9350e5f215ba85"}]}