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
>
>
