The feature works as expected in Scala/Java, but not implemented in Python.
On Thu, Feb 12, 2015 at 9:24 AM, Imran Rashid <iras...@cloudera.com> wrote: > I wonder if the issue is that these lines just need to add > preservesPartitioning = true > ? > > https://github.com/apache/spark/blob/master/python/pyspark/join.py#L38 > > I am getting the feeling this is an issue w/ pyspark > > > On Thu, Feb 12, 2015 at 10:43 AM, Imran Rashid <iras...@cloudera.com> wrote: >> >> ah, sorry I am not too familiar w/ pyspark, sorry I missed that part. It >> could be that pyspark doesn't properly support narrow dependencies, or maybe >> you need to be more explicit about the partitioner. I am looking into the >> pyspark api but you might have some better guesses here than I thought. >> >> My suggestion to do >> >> joinedRdd.getPartitions.foreach{println} >> >> was just to see if the partition was a NarrowCoGroupSplitDep or a >> ShuffleCoGroupSplitDep -- but it was actually a bad suggestion, those fields >> are hidden deeper inside and are not user-visible. But I think a better way >> (in scala, anyway) is to look at rdd.dependencies. its a little tricky, >> though, you need to look deep into the lineage (example at the end). >> >> Sean -- yes it does require both RDDs have the same partitioner, but that >> should happen naturally if you just specify the same number of partitions, >> you'll get equal HashPartitioners. There is a little difference in the >> scala & python api that I missed here. For partitionBy in scala, you >> actually need to specify the partitioner, but not in python. However I >> thought it would work like groupByKey, which does just take an int. >> >> >> Here's a code example in scala -- not sure what is available from python. >> Hopefully somebody knows a simpler way to confirm narrow dependencies?? >> >>> val d = sc.parallelize(1 to 1e6.toInt).map{x => x -> x}.groupByKey(64) >>> val d2 = sc.parallelize(3 to 1e6.toInt).map{x => x -> x}.groupByKey(64) >>> scala> d.partitioner == d2.partitioner >>> res2: Boolean = true >>> val joined = d.join(d2) >>> val d3 = sc.parallelize(3 to 1e6.toInt).map{x => x -> x}.groupByKey(100) >>> val badJoined = d.join(d3) >>> >>> d.setName("d") >>> d2.setName("d2") >>> d3.setName("d3") >>> joined.setName("joined") >>> badJoined.setName("badJoined") >>> >>> >>> //unfortunatley, just looking at the immediate dependencies of joined & >>> badJoined is misleading, b/c join actually creates >>> // one more step after the shuffle >>> scala> joined.dependencies >>> res20: Seq[org.apache.spark.Dependency[_]] = >>> List(org.apache.spark.OneToOneDependency@74751ac8) >>> //even with the join that does require a shuffle, we still see a >>> OneToOneDependency, but thats just a simple flatMap step >>> scala> badJoined.dependencies >>> res21: Seq[org.apache.spark.Dependency[_]] = >>> List(org.apache.spark.OneToOneDependency@1cf356cc) >> >> >> >>> >>> //so lets make a helper function to get all the dependencies recursively >>> >>> def flattenDeps(rdd: RDD[_]): Seq[(RDD[_], Dependency[_])] = { >>> val deps = rdd.dependencies >>> deps.map{rdd -> _} ++ deps.flatMap{dep => flattenDeps(dep.rdd)} >>> } >>> >>> >>> //full dependencies of the good join >>> >>> scala> flattenDeps(joined).foreach{println} >>> (joined FlatMappedValuesRDD[9] at join at >>> <console>:16,org.apache.spark.OneToOneDependency@74751ac8) >>> (MappedValuesRDD[8] at join at >>> <console>:16,org.apache.spark.OneToOneDependency@623264af) >>> (CoGroupedRDD[7] at join at >>> <console>:16,org.apache.spark.OneToOneDependency@5a704f86) >>> (CoGroupedRDD[7] at join at >>> <console>:16,org.apache.spark.OneToOneDependency@37514cd) >>> (d ShuffledRDD[3] at groupByKey at >>> <console>:12,org.apache.spark.ShuffleDependency@7ba8a080) >>> (MappedRDD[2] at map at >>> <console>:12,org.apache.spark.OneToOneDependency@7bc172ec) >>> (d2 ShuffledRDD[6] at groupByKey at >>> <console>:12,org.apache.spark.ShuffleDependency@5960236d) >>> (MappedRDD[5] at map at >>> <console>:12,org.apache.spark.OneToOneDependency@36b5f6f2) >>> >>> >>> >>> //full dependencies of the bad join -- notice the ShuffleDependency! >>> >>> scala> flattenDeps(badJoined).foreach{println} >>> (badJoined FlatMappedValuesRDD[15] at join at >>> <console>:16,org.apache.spark.OneToOneDependency@1cf356cc) >>> (MappedValuesRDD[14] at join at >>> <console>:16,org.apache.spark.OneToOneDependency@5dea4db) >>> (CoGroupedRDD[13] at join at >>> <console>:16,org.apache.spark.ShuffleDependency@5c1928df) >>> (CoGroupedRDD[13] at join at >>> <console>:16,org.apache.spark.OneToOneDependency@77ca77b5) >>> (d ShuffledRDD[3] at groupByKey at >>> <console>:12,org.apache.spark.ShuffleDependency@7ba8a080) >>> (MappedRDD[2] at map at >>> <console>:12,org.apache.spark.OneToOneDependency@7bc172ec) >>> (d3 ShuffledRDD[12] at groupByKey at >>> <console>:12,org.apache.spark.ShuffleDependency@d794984) >>> (MappedRDD[11] at map at >>> <console>:12,org.apache.spark.OneToOneDependency@15c98005) >> >> >> >> On Thu, Feb 12, 2015 at 10:05 AM, Karlson <ksonsp...@siberie.de> wrote: >>> >>> Hi Imran, >>> >>> thanks for your quick reply. >>> >>> Actually I am doing this: >>> >>> rddA = rddA.partitionBy(n).cache() >>> rddB = rddB.partitionBy(n).cache() >>> >>> followed by >>> >>> rddA.count() >>> rddB.count() >>> >>> then joinedRDD = rddA.join(rddB) >>> >>> I thought that the count() would force the evaluation, so any subsequent >>> joins would be shuffleless. I was wrong about the shuffle amounts however. >>> The shuffle write is actually 2GB (i.e. the size of the bigger RDD) whil >>> there is no Shuffle read. A joinedRdd.count() does a shuffle read of about >>> 1GB in size, though. >>> >>> The getPartitions-method does not exist on the resulting RDD (I am using >>> the Python API). There is however foreachPartition(). What is the line >>> >>> joinedRdd.getPartitions.foreach{println} >>> >>> supposed to do? >>> >>> Thank you, >>> >>> Karlson >>> >>> PS: Sorry for sending this twice, I accidentally did not reply to the >>> mailing list first. >>> >>> >>> >>> On 2015-02-12 16:48, Imran Rashid wrote: >>>> >>>> Hi Karlson, >>>> >>>> I think your assumptions are correct -- that join alone shouldn't >>>> require >>>> any shuffling. But its possible you are getting tripped up by lazy >>>> evaluation of RDDs. After you do your partitionBy, are you sure those >>>> RDDs >>>> are actually materialized & cached somewhere? eg., if you just did >>>> this: >>>> >>>> val rddA = someData.partitionBy(N) >>>> val rddB = someOtherData.partitionBy(N) >>>> val joinedRdd = rddA.join(rddB) >>>> joinedRdd.count() //or any other action >>>> >>>> then the partitioning isn't actually getting run until you do the join. >>>> So >>>> though the join itself can happen without partitioning, >>>> joinedRdd.count() >>>> will trigger the evaluation of rddA & rddB which will require shuffles. >>>> Note that even if you have some intervening action on rddA & rddB that >>>> shuffles them, unless you persist the result, you will need to reshuffle >>>> them for the join. >>>> >>>> If this doesn't help explain things, for debugging >>>> >>>> joinedRdd.getPartitions.foreach{println} >>>> >>>> this is getting into the weeds, but at least this will tell us whether >>>> or >>>> not you are getting narrow dependencies, which would avoid the shuffle. >>>> (Does anyone know of a simpler way to check this?) >>>> >>>> hope this helps, >>>> Imran >>>> >>>> >>>> >>>> >>>> On Thu, Feb 12, 2015 at 9:25 AM, Karlson <ksonsp...@siberie.de> wrote: >>>> >>>>> Hi All, >>>>> >>>>> using Pyspark, I create two RDDs (one with about 2M records (~200MB), >>>>> the >>>>> other with about 8M records (~2GB)) of the format (key, value). >>>>> >>>>> I've done a partitionBy(num_partitions) on both RDDs and verified that >>>>> both RDDs have the same number of partitions and that equal keys reside >>>>> on >>>>> the same partition (via mapPartitionsWithIndex). >>>>> >>>>> Now I'd expect that for a join on the two RDDs no shuffling is >>>>> necessary. >>>>> Looking at the Web UI under http://driver:4040 however reveals that >>>>> that >>>>> assumption is false. >>>>> >>>>> In fact I am seeing shuffle writes of about 200MB and reads of about >>>>> 50MB. >>>>> >>>>> What's the explanation for that behaviour? Where am I wrong with my >>>>> assumption? >>>>> >>>>> Thanks in advance, >>>>> >>>>> Karlson >>>>> >>>>> --------------------------------------------------------------------- >>>>> To unsubscribe, e-mail: user-unsubscr...@spark.apache.org >>>>> For additional commands, e-mail: user-h...@spark.apache.org >>>>> >>>>> >>> >>> --------------------------------------------------------------------- >>> To unsubscribe, e-mail: user-unsubscr...@spark.apache.org >>> For additional commands, e-mail: user-h...@spark.apache.org >>> >> > --------------------------------------------------------------------- To unsubscribe, e-mail: user-unsubscr...@spark.apache.org For additional commands, e-mail: user-h...@spark.apache.org
