diff --git a/python/pyspark/sql.py b/python/pyspark/sql.py index 2a09c02c6a3b4..1990323249cf6 100644 --- a/python/pyspark/sql.py +++ b/python/pyspark/sql.py @@ -2039,14 +2039,14 @@ def getCheckpointFile(self): return checkpointFile.get() def coalesce(self, numPartitions, shuffle=False): - rdd = self._jschema_rdd.coalesce(numPartitions, shuffle) + rdd = self._jschema_rdd.coalesce(numPartitions, shuffle, None) return SchemaRDD(rdd, self.sql_ctx) def distinct(self, numPartitions=None): if numPartitions is None: rdd = self._jschema_rdd.distinct() else: - rdd = self._jschema_rdd.distinct(numPartitions) + rdd = self._jschema_rdd.distinct(numPartitions, None) return SchemaRDD(rdd, self.sql_ctx) def intersection(self, other): @@ -2057,7 +2057,7 @@ def intersection(self, other): raise ValueError("Can only intersect with another SchemaRDD") def repartition(self, numPartitions): - rdd = self._jschema_rdd.repartition(numPartitions) + rdd = self._jschema_rdd.repartition(numPartitions, None) return SchemaRDD(rdd, self.sql_ctx) def subtract(self, other, numPartitions=None): diff --git a/sql/core/src/main/scala/org/apache/spark/sql/SchemaRDD.scala b/sql/core/src/main/scala/org/apache/spark/sql/SchemaRDD.scala index 9062c9c107d33..ae4d8ba90c5bd 100644 --- a/sql/core/src/main/scala/org/apache/spark/sql/SchemaRDD.scala +++ b/sql/core/src/main/scala/org/apache/spark/sql/SchemaRDD.scala @@ -476,9 +476,6 @@ class SchemaRDD( (implicit ord: Ordering[Row] = null): SchemaRDD = applySchema(super.coalesce(numPartitions, shuffle)(ord)) - def coalesce(numPartitions: Int, shuffle: Boolean): SchemaRDD = - applySchema(super.coalesce(numPartitions, shuffle)(null)) - override def distinct(): SchemaRDD = applySchema(super.distinct()) override def distinct(numPartitions: Int) @@ -505,9 +502,6 @@ class SchemaRDD( (implicit ord: Ordering[Row] = null): SchemaRDD = applySchema(super.repartition(numPartitions)(ord)) - def repartition(numPartitions: Int): SchemaRDD = - applySchema(super.repartition(numPartitions)(null)) - override def subtract(other: RDD[Row]): SchemaRDD = applySchema(super.subtract(other))