diff options
author | Josh Rosen <joshrosen@databricks.com> | 2015-07-22 21:04:04 -0700 |
---|---|---|
committer | Reynold Xin <rxin@databricks.com> | 2015-07-22 21:04:04 -0700 |
commit | b217230f2a96c6d5a0554c593bdf1d1374878688 (patch) | |
tree | 88033457313049d3b079cc45a32bed8adc59737e /python/pyspark | |
parent | d71a13f475df2d05a7db9e25738d1353cbc8cfc7 (diff) | |
download | spark-b217230f2a96c6d5a0554c593bdf1d1374878688.tar.gz spark-b217230f2a96c6d5a0554c593bdf1d1374878688.tar.bz2 spark-b217230f2a96c6d5a0554c593bdf1d1374878688.zip |
[SPARK-9144] Remove DAGScheduler.runLocallyWithinThread and spark.localExecution.enabled
Spark has an option called spark.localExecution.enabled; according to the docs:
> Enables Spark to run certain jobs, such as first() or take() on the driver, without sending tasks to the cluster. This can make certain jobs execute very quickly, but may require shipping a whole partition of data to the driver.
This feature ends up adding quite a bit of complexity to DAGScheduler, especially in the runLocallyWithinThread method, but as far as I know nobody uses this feature (I searched the mailing list and haven't seen any recent mentions of the configuration nor stacktraces including the runLocally method). As a step towards scheduler complexity reduction, I propose that we remove this feature and all code related to it for Spark 1.5.
This pull request simply brings #7484 up to date.
Author: Josh Rosen <joshrosen@databricks.com>
Author: Reynold Xin <rxin@databricks.com>
Closes #7585 from rxin/remove-local-exec and squashes the following commits:
84bd10e [Reynold Xin] Python fix.
1d9739a [Reynold Xin] Merge pull request #7484 from JoshRosen/remove-localexecution
eec39fa [Josh Rosen] Remove allowLocal(); deprecate user-facing uses of it.
b0835dc [Josh Rosen] Remove local execution code in DAGScheduler
8975d96 [Josh Rosen] Remove local execution tests.
ffa8c9b [Josh Rosen] Remove documentation for configuration
Diffstat (limited to 'python/pyspark')
-rw-r--r-- | python/pyspark/context.py | 3 | ||||
-rw-r--r-- | python/pyspark/rdd.py | 4 |
2 files changed, 3 insertions, 4 deletions
diff --git a/python/pyspark/context.py b/python/pyspark/context.py index 43bde5ae41..eb5b0bbbda 100644 --- a/python/pyspark/context.py +++ b/python/pyspark/context.py @@ -913,8 +913,7 @@ class SparkContext(object): # by runJob() in order to avoid having to pass a Python lambda into # SparkContext#runJob. mappedRDD = rdd.mapPartitions(partitionFunc) - port = self._jvm.PythonRDD.runJob(self._jsc.sc(), mappedRDD._jrdd, partitions, - allowLocal) + port = self._jvm.PythonRDD.runJob(self._jsc.sc(), mappedRDD._jrdd, partitions) return list(_load_from_socket(port, mappedRDD._jrdd_deserializer)) def show_profiles(self): diff --git a/python/pyspark/rdd.py b/python/pyspark/rdd.py index 7e788148d9..fa8e0a0574 100644 --- a/python/pyspark/rdd.py +++ b/python/pyspark/rdd.py @@ -1293,7 +1293,7 @@ class RDD(object): taken += 1 p = range(partsScanned, min(partsScanned + numPartsToTry, totalParts)) - res = self.context.runJob(self, takeUpToNumLeft, p, True) + res = self.context.runJob(self, takeUpToNumLeft, p) items += res partsScanned += numPartsToTry @@ -2193,7 +2193,7 @@ class RDD(object): values = self.filter(lambda kv: kv[0] == key).values() if self.partitioner is not None: - return self.ctx.runJob(values, lambda x: x, [self.partitioner(key)], False) + return self.ctx.runJob(values, lambda x: x, [self.partitioner(key)]) return values.collect() |