Hi Imran,
This is extremely helpful. This is not only an approach, also help me to
understand how to affect or customize my own DAG effectively.
Thanks a lot!
Shuai
On Monday, March 16, 2015, Imran Rashid wrote:
> Hi Shuai,
>
> yup, that is exactly what I meant -- implement your own class
> MyGroupingRDD. This is definitely more detail than a lot of users will
> need to go, but its also not all that scary either. In this case, you want
> something that is *extremely* close to the existing CoalescedRDD, so start
> by looking at that code.
>
>
> https://github.com/apache/spark/blob/master/core/src/main/scala/org/apache/spark/rdd/CoalescedRDD.scala
>
> The only thing which is complicated in CoalescedRDD is the
> PartitionCoalescer, but that is completely irrelevant for you, so you can
> ignore it. I started writing up a description of what to do but then I
> realized just writing the code would be easier :) Totally untested, but
> here you go:
>
> https://gist.github.com/squito/c2d1dd5413a60830d6f3
>
> The only really interesting part here is getPartitions:
>
>
> https://gist.github.com/squito/c2d1dd5413a60830d6f3#file-groupedrdd-scala-L31
>
> That's where you create partitions in your new RDD, which depend on
> multiple RDDs from the parent. Also note that compute() is very simple:
> you just concatenate together the iterators from each of the parent RDDs:
>
>
> https://gist.github.com/squito/c2d1dd5413a60830d6f3#file-groupedrdd-scala-L37
>
> let me know how it goes!
>
>
> On Mon, Mar 16, 2015 at 5:15 PM, Shuai Zheng > wrote:
>
>> Hi Imran,
>>
>>
>>
>> I am a bit confused here. Assume I have RDD a with 1000 partition and
>> also has been sorted. How can I control when creating RDD b (with 20
>> partitions) to make sure 1-50 partition of RDD a map to 1st partition of
>> RDD b? I don’t see any control code/logic here?
>>
>>
>>
>> You code below:
>>
>>
>>
>> val groupedRawData20Partitions = new MyGroupingRDD(rawData1000Partitions)
>>
>>
>>
>>
>>
>> Does it means I need to define/develop my own MyGroupingRDD class? I am
>> not very clear how to do that, any place I can find an example? I never
>> create my own RDD class before (not RDD instance J). But this is very
>> valuable approach to me so I am desired to learn.
>>
>>
>>
>> Regards,
>>
>>
>>
>> Shuai
>>
>>
>>
>> *From:* Imran Rashid [mailto:iras...@cloudera.com
>> ]
>> *Sent:* Monday, March 16, 2015 11:22 AM
>> *To:* Shawn Zheng; user@spark.apache.org
>>
>> *Subject:* Re: Process time series RDD after sortByKey
>>
>>
>>
>> Hi Shuai,
>>
>>
>>
>> On Sat, Mar 14, 2015 at 11:02 AM, Shawn Zheng > > wrote:
>>
>> Sorry I response late.
>>
>> Zhan Zhang's solution is very interesting and I look at into it, but it
>> is not what I want. Basically I want to run the job sequentially and also
>> gain parallelism. So if possible, if I have 1000 partition, the best case
>> is I can run it as 20 subtask, each one take partition: 1-50, 51-100,
>> 101-150, etc.
>>
>> If we have ability to do this, we will gain huge flexibility when we try
>> to process some time series like data and a lot of algo will benefit from
>> it.
>>
>>
>>
>> yes, this is what I was suggesting you do. You would first create one
>> RDD (a) that has 1000 partitions. Don't worry about the creation of this
>> RDD -- it wont' create any tasks, its just a logical holder of your raw
>> data. Then you create another RDD (b) that depends on your RDD (a), but
>> that only has 20 partitions. Each partition in (b) would depend on a
>> number of partitions from (a). As you've suggested, partition 1 in (b)
>> would depend on partitions 1-50 in (a), partition 2 in (b) would depend on
>> 51-100 in (a), etc. Note that RDD (b) still doesn't *do* anything. Its
>> just another logical holder for your data, but this time grouped in the way
>> you want. Then after RDD (b), you would do whatever other transformations
>> you wanted, but now you'd be working w/ 20 partitions:
>>
>>
>>
>> val rawData1000Partitions = sc.textFile(...) // or whatever
>>
>> val groupedRawData20Partitions = new MyGroupingRDD(rawData1000Partitions)
>>
>> groupedRawData20Partitions.map{...}.filter{...}.reduceByKey{...} //etc.
>>
>>
>>
>> note that this is almost exactly the same as what CoalescedRdd does.
>> However, it might combine the
