You are viewing a plain text version of this content. The canonical link for it is here.
Posted to user@spark.apache.org by Maciej Sokołowski <ma...@gmail.com> on 2016/05/31 13:13:52 UTC
Re: Splitting RDD to exact number of partitions
Hello Spark users and developers.
I read file and want to ensure that it has exact number of partitions, for
example 128.
In documentation I found:
def textFile(path: String, minPartitions: Int = defaultMinPartitions):
RDD[String]
But argument here is minimal number of partitions, so I use coalesce to
ensure desired number of partitions:
def coalesce(numPartitions: Int, shuffle: Boolean = false)(implicit ord:
Ordering[T] = null): RDD[T]
//Return a new RDD that is reduced into numPartitions partitions.
So I combine them and get number of partitions lower than expected:
scala> sc.textFile("perf_test1.csv",
minPartitions=128).coalesce(128).getNumPartitions
res14: Int = 126
Is this expected behaviour? File contains 100000 lines, size of partitions
before and after coalesce:
scala> sc.textFile("perf_test1.csv", minPartitions=128).mapPartitions{rows
=> Iterator(rows.length)}.collect()
res16: Array[Int] = Array(782, 781, 782, 781, 781, 782, 781, 781, 781, 781,
782, 781, 781, 781, 782, 781, 781, 781, 782, 781, 781, 781, 782, 781, 781,
781, 782, 781, 781, 782, 781, 781, 781, 781, 782, 781, 781, 781, 782, 781,
781, 781, 782, 781, 781, 782, 781, 781, 782, 781, 781, 781, 781, 782, 781,
781, 781, 782, 781, 781, 781, 782, 781, 781, 781, 781, 782, 781, 781, 782,
781, 781, 782, 781, 781, 781, 781, 782, 781, 781, 781, 782, 781, 781, 781,
782, 781, 781, 782, 781, 781, 781, 782, 781, 781, 781, 781, 782, 781, 781,
781, 781, 781, 782, 781, 781, 781, 782, 781, 781, 782, 781, 781, 781, 781,
782, 781, 781, 781, 781, 782, 781, 781, 782, 781, 781, 781, 781)
scala> sc.textFile("perf_test1.csv",
minPartitions=128).coalesce(128).mapPartitions{rows =>
Iterator(rows.length)}.collect()
res15: Array[Int] = Array(1563, 781, 781, 781, 782, 781, 781, 781, 781,
782, 781, 781, 781, 781, 782, 781, 781, 781, 781, 781, 782, 781, 781, 781,
782, 781, 781, 782, 781, 781, 781, 781, 782, 781, 781, 781, 781, 782, 781,
781, 782, 781, 781, 781, 781, 1563, 782, 781, 781, 782, 781, 781, 781, 781,
782, 781, 781, 781, 782, 781, 781, 781, 782, 781, 781, 781, 782, 781, 781,
781, 782, 781, 781, 782, 781, 781, 781, 781, 782, 781, 781, 781, 782, 781,
781, 781, 782, 781, 781, 782, 781, 781, 782, 781, 781, 781, 781, 782, 781,
781, 781, 782, 781, 781, 781, 782, 781, 781, 781, 781, 782, 781, 781, 782,
781, 781, 782, 781, 781, 781, 781, 782, 781, 781, 781, 782)
So two partitions are double the size. Is this expected behaviour or is it
some kind of bug?
Thanks,
Maciej Sokołowski
Re: Splitting RDD to exact number of partitions
Posted by Maciej Sokołowski <ma...@gmail.com>.
Thanks.
At what conditions number of partitions can be higher than minPartitions
when reading textFile? Should this be considered as unfrequent situation?
To sum up - is there more efficient way to ensure exact number of
partitions than following:
rdd = sc.textFile("perf_test1.csv", minPartitions=128)
if (rdd.getNumPartitions > 128)
rdd.repartition(128)
?
On 31 May 2016 at 16:11, Takeshi Yamamuro <li...@gmail.com> wrote:
> `coalesce` without shuffling can only set fewer partitions than its
> parent RDD.
> As Ted said, you need to set true in shuffle, or you need to use
> `RDD#repartition`.
>
> // maropu
>
> On Tue, May 31, 2016 at 11:02 PM, Ted Yu <yu...@gmail.com> wrote:
>
>> Value for shuffle is false by default.
>>
>> Have you tried setting it to true ?
>>
>> Which Spark release are you using ?
>>
>> On Tue, May 31, 2016 at 6:13 AM, Maciej Sokołowski <ma...@gmail.com>
>> wrote:
>>
>>> Hello Spark users and developers.
>>>
>>> I read file and want to ensure that it has exact number of partitions,
>>> for example 128.
>>>
>>> In documentation I found:
>>>
>>> def textFile(path: String, minPartitions: Int = defaultMinPartitions):
>>> RDD[String]
>>>
>>> But argument here is minimal number of partitions, so I use coalesce to
>>> ensure desired number of partitions:
>>>
>>> def coalesce(numPartitions: Int, shuffle: Boolean = false)(implicit ord:
>>> Ordering[T] = null): RDD[T]
>>> //Return a new RDD that is reduced into numPartitions partitions.
>>>
>>> So I combine them and get number of partitions lower than expected:
>>>
>>> scala> sc.textFile("perf_test1.csv",
>>> minPartitions=128).coalesce(128).getNumPartitions
>>> res14: Int = 126
>>>
>>> Is this expected behaviour? File contains 100000 lines, size of
>>> partitions before and after coalesce:
>>>
>>> scala> sc.textFile("perf_test1.csv",
>>> minPartitions=128).mapPartitions{rows => Iterator(rows.length)}.collect()
>>> res16: Array[Int] = Array(782, 781, 782, 781, 781, 782, 781, 781, 781,
>>> 781, 782, 781, 781, 781, 782, 781, 781, 781, 782, 781, 781, 781, 782, 781,
>>> 781, 781, 782, 781, 781, 782, 781, 781, 781, 781, 782, 781, 781, 781, 782,
>>> 781, 781, 781, 782, 781, 781, 782, 781, 781, 782, 781, 781, 781, 781, 782,
>>> 781, 781, 781, 782, 781, 781, 781, 782, 781, 781, 781, 781, 782, 781, 781,
>>> 782, 781, 781, 782, 781, 781, 781, 781, 782, 781, 781, 781, 782, 781, 781,
>>> 781, 782, 781, 781, 782, 781, 781, 781, 782, 781, 781, 781, 781, 782, 781,
>>> 781, 781, 781, 781, 782, 781, 781, 781, 782, 781, 781, 782, 781, 781, 781,
>>> 781, 782, 781, 781, 781, 781, 782, 781, 781, 782, 781, 781, 781, 781)
>>>
>>> scala> sc.textFile("perf_test1.csv",
>>> minPartitions=128).coalesce(128).mapPartitions{rows =>
>>> Iterator(rows.length)}.collect()
>>> res15: Array[Int] = Array(1563, 781, 781, 781, 782, 781, 781, 781, 781,
>>> 782, 781, 781, 781, 781, 782, 781, 781, 781, 781, 781, 782, 781, 781, 781,
>>> 782, 781, 781, 782, 781, 781, 781, 781, 782, 781, 781, 781, 781, 782, 781,
>>> 781, 782, 781, 781, 781, 781, 1563, 782, 781, 781, 782, 781, 781, 781, 781,
>>> 782, 781, 781, 781, 782, 781, 781, 781, 782, 781, 781, 781, 782, 781, 781,
>>> 781, 782, 781, 781, 782, 781, 781, 781, 781, 782, 781, 781, 781, 782, 781,
>>> 781, 781, 782, 781, 781, 782, 781, 781, 782, 781, 781, 781, 781, 782, 781,
>>> 781, 781, 782, 781, 781, 781, 782, 781, 781, 781, 781, 782, 781, 781, 782,
>>> 781, 781, 782, 781, 781, 781, 781, 782, 781, 781, 781, 782)
>>>
>>> So two partitions are double the size. Is this expected behaviour or is
>>> it some kind of bug?
>>>
>>> Thanks,
>>> Maciej Sokołowski
>>>
>>
>>
>
>
> --
> ---
> Takeshi Yamamuro
>
Re: Splitting RDD to exact number of partitions
Posted by Takeshi Yamamuro <li...@gmail.com>.
`coalesce` without shuffling can only set fewer partitions than its parent
RDD.
As Ted said, you need to set true in shuffle, or you need to use
`RDD#repartition`.
// maropu
On Tue, May 31, 2016 at 11:02 PM, Ted Yu <yu...@gmail.com> wrote:
> Value for shuffle is false by default.
>
> Have you tried setting it to true ?
>
> Which Spark release are you using ?
>
> On Tue, May 31, 2016 at 6:13 AM, Maciej Sokołowski <ma...@gmail.com>
> wrote:
>
>> Hello Spark users and developers.
>>
>> I read file and want to ensure that it has exact number of partitions,
>> for example 128.
>>
>> In documentation I found:
>>
>> def textFile(path: String, minPartitions: Int = defaultMinPartitions):
>> RDD[String]
>>
>> But argument here is minimal number of partitions, so I use coalesce to
>> ensure desired number of partitions:
>>
>> def coalesce(numPartitions: Int, shuffle: Boolean = false)(implicit ord:
>> Ordering[T] = null): RDD[T]
>> //Return a new RDD that is reduced into numPartitions partitions.
>>
>> So I combine them and get number of partitions lower than expected:
>>
>> scala> sc.textFile("perf_test1.csv",
>> minPartitions=128).coalesce(128).getNumPartitions
>> res14: Int = 126
>>
>> Is this expected behaviour? File contains 100000 lines, size of
>> partitions before and after coalesce:
>>
>> scala> sc.textFile("perf_test1.csv",
>> minPartitions=128).mapPartitions{rows => Iterator(rows.length)}.collect()
>> res16: Array[Int] = Array(782, 781, 782, 781, 781, 782, 781, 781, 781,
>> 781, 782, 781, 781, 781, 782, 781, 781, 781, 782, 781, 781, 781, 782, 781,
>> 781, 781, 782, 781, 781, 782, 781, 781, 781, 781, 782, 781, 781, 781, 782,
>> 781, 781, 781, 782, 781, 781, 782, 781, 781, 782, 781, 781, 781, 781, 782,
>> 781, 781, 781, 782, 781, 781, 781, 782, 781, 781, 781, 781, 782, 781, 781,
>> 782, 781, 781, 782, 781, 781, 781, 781, 782, 781, 781, 781, 782, 781, 781,
>> 781, 782, 781, 781, 782, 781, 781, 781, 782, 781, 781, 781, 781, 782, 781,
>> 781, 781, 781, 781, 782, 781, 781, 781, 782, 781, 781, 782, 781, 781, 781,
>> 781, 782, 781, 781, 781, 781, 782, 781, 781, 782, 781, 781, 781, 781)
>>
>> scala> sc.textFile("perf_test1.csv",
>> minPartitions=128).coalesce(128).mapPartitions{rows =>
>> Iterator(rows.length)}.collect()
>> res15: Array[Int] = Array(1563, 781, 781, 781, 782, 781, 781, 781, 781,
>> 782, 781, 781, 781, 781, 782, 781, 781, 781, 781, 781, 782, 781, 781, 781,
>> 782, 781, 781, 782, 781, 781, 781, 781, 782, 781, 781, 781, 781, 782, 781,
>> 781, 782, 781, 781, 781, 781, 1563, 782, 781, 781, 782, 781, 781, 781, 781,
>> 782, 781, 781, 781, 782, 781, 781, 781, 782, 781, 781, 781, 782, 781, 781,
>> 781, 782, 781, 781, 782, 781, 781, 781, 781, 782, 781, 781, 781, 782, 781,
>> 781, 781, 782, 781, 781, 782, 781, 781, 782, 781, 781, 781, 781, 782, 781,
>> 781, 781, 782, 781, 781, 781, 782, 781, 781, 781, 781, 782, 781, 781, 782,
>> 781, 781, 782, 781, 781, 781, 781, 782, 781, 781, 781, 782)
>>
>> So two partitions are double the size. Is this expected behaviour or is
>> it some kind of bug?
>>
>> Thanks,
>> Maciej Sokołowski
>>
>
>
--
---
Takeshi Yamamuro
Re: Splitting RDD to exact number of partitions
Posted by Ovidiu-Cristian MARCU <ov...@inria.fr>.
Hi Ted,
Any chance to develop more on the SQLConf parameters in the sense to have more explanations for changing these settings?
Not all of them are made clear in the descriptions.
Thanks!
Best,
Ovidiu
> On 31 May 2016, at 16:30, Ted Yu <yu...@gmail.com> wrote:
>
> Maciej:
> You can refer to the doc in sql/core/src/main/scala/org/apache/spark/sql/internal/SQLConf.scala for these parameters.
>
> On Tue, May 31, 2016 at 7:27 AM, Takeshi Yamamuro <linguin.m.s@gmail.com <ma...@gmail.com>> wrote:
> If you don't hesitate the newest version, you try to use v2.0-preview.
> http://spark.apache.org/news/spark-2.0.0-preview.html <http://spark.apache.org/news/spark-2.0.0-preview.html>
>
> There, you can control #partitions for input partitions without shuffles by two parameters below;
> spark.sql.files.maxPartitionBytes
> spark.sql.files.openCostInBytes
> ( Not documented though, ....
>
> // maropu
>
> On Tue, May 31, 2016 at 11:08 PM, Maciej Sokołowski <matemaciek@gmail.com <ma...@gmail.com>> wrote:
> After setting shuffle to true I get expected 128 partitions, but I'm worried about performance of such solution - especially I see that some shuffling is done because size of partitions chages:
>
> scala> sc.textFile("hdfs:///proj/dFAB_test/testdata/perf_test1.csv", minPartitions=128).coalesce(128, true).mapPartitions{rows => Iterator(rows.length)}.collect()
> res3: Array[Int] = Array(768, 768, 768, 768, 768, 768, 768, 768, 768, 768, 768, 768, 768, 768, 768, 768, 768, 768, 768, 768, 768, 768, 768, 768, 768, 768, 768, 768, 768, 768, 768, 768, 768, 768, 768, 768, 768, 768, 768, 768, 768, 768, 768, 768, 768, 768, 768, 768, 768, 768, 768, 768, 768, 768, 768, 768, 768, 768, 768, 768, 768, 768, 768, 768, 768, 768, 768, 768, 768, 768, 768, 768, 768, 768, 768, 768, 768, 768, 768, 768, 768, 768, 768, 768, 768, 768, 768, 768, 768, 768, 768, 768, 768, 828, 896, 896, 896, 896, 896, 896, 896, 896, 896, 896, 896, 896, 850, 786, 768, 768, 768, 768, 768, 768, 768, 768, 768, 768, 768, 768, 768, 768, 768, 768, 768, 768, 768, 768)
>
> I use spark 1.6.0
>
>
> On 31 May 2016 at 16:02, Ted Yu <yuzhihong@gmail.com <ma...@gmail.com>> wrote:
> Value for shuffle is false by default.
>
> Have you tried setting it to true ?
>
> Which Spark release are you using ?
>
> On Tue, May 31, 2016 at 6:13 AM, Maciej Sokołowski <matemaciek@gmail.com <ma...@gmail.com>> wrote:
> Hello Spark users and developers.
>
> I read file and want to ensure that it has exact number of partitions, for example 128.
>
> In documentation I found:
>
> def textFile(path: String, minPartitions: Int = defaultMinPartitions): RDD[String]
>
> But argument here is minimal number of partitions, so I use coalesce to ensure desired number of partitions:
>
> def coalesce(numPartitions: Int, shuffle: Boolean = false)(implicit ord: Ordering[T] = null): RDD[T]
> //Return a new RDD that is reduced into numPartitions partitions.
>
> So I combine them and get number of partitions lower than expected:
>
> scala> sc.textFile("perf_test1.csv", minPartitions=128).coalesce(128).getNumPartitions
> res14: Int = 126
>
> Is this expected behaviour? File contains 100000 lines, size of partitions before and after coalesce:
>
> scala> sc.textFile("perf_test1.csv", minPartitions=128).mapPartitions{rows => Iterator(rows.length)}.collect()
> res16: Array[Int] = Array(782, 781, 782, 781, 781, 782, 781, 781, 781, 781, 782, 781, 781, 781, 782, 781, 781, 781, 782, 781, 781, 781, 782, 781, 781, 781, 782, 781, 781, 782, 781, 781, 781, 781, 782, 781, 781, 781, 782, 781, 781, 781, 782, 781, 781, 782, 781, 781, 782, 781, 781, 781, 781, 782, 781, 781, 781, 782, 781, 781, 781, 782, 781, 781, 781, 781, 782, 781, 781, 782, 781, 781, 782, 781, 781, 781, 781, 782, 781, 781, 781, 782, 781, 781, 781, 782, 781, 781, 782, 781, 781, 781, 782, 781, 781, 781, 781, 782, 781, 781, 781, 781, 781, 782, 781, 781, 781, 782, 781, 781, 782, 781, 781, 781, 781, 782, 781, 781, 781, 781, 782, 781, 781, 782, 781, 781, 781, 781)
>
> scala> sc.textFile("perf_test1.csv", minPartitions=128).coalesce(128).mapPartitions{rows => Iterator(rows.length)}.collect()
> res15: Array[Int] = Array(1563, 781, 781, 781, 782, 781, 781, 781, 781, 782, 781, 781, 781, 781, 782, 781, 781, 781, 781, 781, 782, 781, 781, 781, 782, 781, 781, 782, 781, 781, 781, 781, 782, 781, 781, 781, 781, 782, 781, 781, 782, 781, 781, 781, 781, 1563, 782, 781, 781, 782, 781, 781, 781, 781, 782, 781, 781, 781, 782, 781, 781, 781, 782, 781, 781, 781, 782, 781, 781, 781, 782, 781, 781, 782, 781, 781, 781, 781, 782, 781, 781, 781, 782, 781, 781, 781, 782, 781, 781, 782, 781, 781, 782, 781, 781, 781, 781, 782, 781, 781, 781, 782, 781, 781, 781, 782, 781, 781, 781, 781, 782, 781, 781, 782, 781, 781, 782, 781, 781, 781, 781, 782, 781, 781, 781, 782)
>
> So two partitions are double the size. Is this expected behaviour or is it some kind of bug?
>
> Thanks,
> Maciej Sokołowski
>
>
>
>
>
> --
> ---
> Takeshi Yamamuro
>
Re: Splitting RDD to exact number of partitions
Posted by Ted Yu <yu...@gmail.com>.
Maciej:
You can refer to the doc in
sql/core/src/main/scala/org/apache/spark/sql/internal/SQLConf.scala
for these parameters.
On Tue, May 31, 2016 at 7:27 AM, Takeshi Yamamuro <li...@gmail.com>
wrote:
> If you don't hesitate the newest version, you try to use v2.0-preview.
> http://spark.apache.org/news/spark-2.0.0-preview.html
>
> There, you can control #partitions for input partitions without shuffles
> by two parameters below;
> spark.sql.files.maxPartitionBytes
> spark.sql.files.openCostInBytes
> ( Not documented though, ....
>
> // maropu
>
> On Tue, May 31, 2016 at 11:08 PM, Maciej Sokołowski <ma...@gmail.com>
> wrote:
>
>> After setting shuffle to true I get expected 128 partitions, but I'm
>> worried about performance of such solution - especially I see that some
>> shuffling is done because size of partitions chages:
>>
>> scala> sc.textFile("hdfs:///proj/dFAB_test/testdata/perf_test1.csv",
>> minPartitions=128).coalesce(128, true).mapPartitions{rows =>
>> Iterator(rows.length)}.collect()
>> res3: Array[Int] = Array(768, 768, 768, 768, 768, 768, 768, 768, 768,
>> 768, 768, 768, 768, 768, 768, 768, 768, 768, 768, 768, 768, 768, 768, 768,
>> 768, 768, 768, 768, 768, 768, 768, 768, 768, 768, 768, 768, 768, 768, 768,
>> 768, 768, 768, 768, 768, 768, 768, 768, 768, 768, 768, 768, 768, 768, 768,
>> 768, 768, 768, 768, 768, 768, 768, 768, 768, 768, 768, 768, 768, 768, 768,
>> 768, 768, 768, 768, 768, 768, 768, 768, 768, 768, 768, 768, 768, 768, 768,
>> 768, 768, 768, 768, 768, 768, 768, 768, 768, 828, 896, 896, 896, 896, 896,
>> 896, 896, 896, 896, 896, 896, 896, 850, 786, 768, 768, 768, 768, 768, 768,
>> 768, 768, 768, 768, 768, 768, 768, 768, 768, 768, 768, 768, 768, 768)
>>
>> I use spark 1.6.0
>>
>>
>> On 31 May 2016 at 16:02, Ted Yu <yu...@gmail.com> wrote:
>>
>>> Value for shuffle is false by default.
>>>
>>> Have you tried setting it to true ?
>>>
>>> Which Spark release are you using ?
>>>
>>> On Tue, May 31, 2016 at 6:13 AM, Maciej Sokołowski <matemaciek@gmail.com
>>> > wrote:
>>>
>>>> Hello Spark users and developers.
>>>>
>>>> I read file and want to ensure that it has exact number of partitions,
>>>> for example 128.
>>>>
>>>> In documentation I found:
>>>>
>>>> def textFile(path: String, minPartitions: Int = defaultMinPartitions):
>>>> RDD[String]
>>>>
>>>> But argument here is minimal number of partitions, so I use coalesce to
>>>> ensure desired number of partitions:
>>>>
>>>> def coalesce(numPartitions: Int, shuffle: Boolean = false)(implicit
>>>> ord: Ordering[T] = null): RDD[T]
>>>> //Return a new RDD that is reduced into numPartitions partitions.
>>>>
>>>> So I combine them and get number of partitions lower than expected:
>>>>
>>>> scala> sc.textFile("perf_test1.csv",
>>>> minPartitions=128).coalesce(128).getNumPartitions
>>>> res14: Int = 126
>>>>
>>>> Is this expected behaviour? File contains 100000 lines, size of
>>>> partitions before and after coalesce:
>>>>
>>>> scala> sc.textFile("perf_test1.csv",
>>>> minPartitions=128).mapPartitions{rows => Iterator(rows.length)}.collect()
>>>> res16: Array[Int] = Array(782, 781, 782, 781, 781, 782, 781, 781, 781,
>>>> 781, 782, 781, 781, 781, 782, 781, 781, 781, 782, 781, 781, 781, 782, 781,
>>>> 781, 781, 782, 781, 781, 782, 781, 781, 781, 781, 782, 781, 781, 781, 782,
>>>> 781, 781, 781, 782, 781, 781, 782, 781, 781, 782, 781, 781, 781, 781, 782,
>>>> 781, 781, 781, 782, 781, 781, 781, 782, 781, 781, 781, 781, 782, 781, 781,
>>>> 782, 781, 781, 782, 781, 781, 781, 781, 782, 781, 781, 781, 782, 781, 781,
>>>> 781, 782, 781, 781, 782, 781, 781, 781, 782, 781, 781, 781, 781, 782, 781,
>>>> 781, 781, 781, 781, 782, 781, 781, 781, 782, 781, 781, 782, 781, 781, 781,
>>>> 781, 782, 781, 781, 781, 781, 782, 781, 781, 782, 781, 781, 781, 781)
>>>>
>>>> scala> sc.textFile("perf_test1.csv",
>>>> minPartitions=128).coalesce(128).mapPartitions{rows =>
>>>> Iterator(rows.length)}.collect()
>>>> res15: Array[Int] = Array(1563, 781, 781, 781, 782, 781, 781, 781, 781,
>>>> 782, 781, 781, 781, 781, 782, 781, 781, 781, 781, 781, 782, 781, 781, 781,
>>>> 782, 781, 781, 782, 781, 781, 781, 781, 782, 781, 781, 781, 781, 782, 781,
>>>> 781, 782, 781, 781, 781, 781, 1563, 782, 781, 781, 782, 781, 781, 781, 781,
>>>> 782, 781, 781, 781, 782, 781, 781, 781, 782, 781, 781, 781, 782, 781, 781,
>>>> 781, 782, 781, 781, 782, 781, 781, 781, 781, 782, 781, 781, 781, 782, 781,
>>>> 781, 781, 782, 781, 781, 782, 781, 781, 782, 781, 781, 781, 781, 782, 781,
>>>> 781, 781, 782, 781, 781, 781, 782, 781, 781, 781, 781, 782, 781, 781, 782,
>>>> 781, 781, 782, 781, 781, 781, 781, 782, 781, 781, 781, 782)
>>>>
>>>> So two partitions are double the size. Is this expected behaviour or is
>>>> it some kind of bug?
>>>>
>>>> Thanks,
>>>> Maciej Sokołowski
>>>>
>>>
>>>
>>
>
>
> --
> ---
> Takeshi Yamamuro
>
Re: Splitting RDD to exact number of partitions
Posted by Takeshi Yamamuro <li...@gmail.com>.
If you don't hesitate the newest version, you try to use v2.0-preview.
http://spark.apache.org/news/spark-2.0.0-preview.html
There, you can control #partitions for input partitions without shuffles by
two parameters below;
spark.sql.files.maxPartitionBytes
spark.sql.files.openCostInBytes
( Not documented though, ....
// maropu
On Tue, May 31, 2016 at 11:08 PM, Maciej Sokołowski <ma...@gmail.com>
wrote:
> After setting shuffle to true I get expected 128 partitions, but I'm
> worried about performance of such solution - especially I see that some
> shuffling is done because size of partitions chages:
>
> scala> sc.textFile("hdfs:///proj/dFAB_test/testdata/perf_test1.csv",
> minPartitions=128).coalesce(128, true).mapPartitions{rows =>
> Iterator(rows.length)}.collect()
> res3: Array[Int] = Array(768, 768, 768, 768, 768, 768, 768, 768, 768, 768,
> 768, 768, 768, 768, 768, 768, 768, 768, 768, 768, 768, 768, 768, 768, 768,
> 768, 768, 768, 768, 768, 768, 768, 768, 768, 768, 768, 768, 768, 768, 768,
> 768, 768, 768, 768, 768, 768, 768, 768, 768, 768, 768, 768, 768, 768, 768,
> 768, 768, 768, 768, 768, 768, 768, 768, 768, 768, 768, 768, 768, 768, 768,
> 768, 768, 768, 768, 768, 768, 768, 768, 768, 768, 768, 768, 768, 768, 768,
> 768, 768, 768, 768, 768, 768, 768, 768, 828, 896, 896, 896, 896, 896, 896,
> 896, 896, 896, 896, 896, 896, 850, 786, 768, 768, 768, 768, 768, 768, 768,
> 768, 768, 768, 768, 768, 768, 768, 768, 768, 768, 768, 768, 768)
>
> I use spark 1.6.0
>
>
> On 31 May 2016 at 16:02, Ted Yu <yu...@gmail.com> wrote:
>
>> Value for shuffle is false by default.
>>
>> Have you tried setting it to true ?
>>
>> Which Spark release are you using ?
>>
>> On Tue, May 31, 2016 at 6:13 AM, Maciej Sokołowski <ma...@gmail.com>
>> wrote:
>>
>>> Hello Spark users and developers.
>>>
>>> I read file and want to ensure that it has exact number of partitions,
>>> for example 128.
>>>
>>> In documentation I found:
>>>
>>> def textFile(path: String, minPartitions: Int = defaultMinPartitions):
>>> RDD[String]
>>>
>>> But argument here is minimal number of partitions, so I use coalesce to
>>> ensure desired number of partitions:
>>>
>>> def coalesce(numPartitions: Int, shuffle: Boolean = false)(implicit ord:
>>> Ordering[T] = null): RDD[T]
>>> //Return a new RDD that is reduced into numPartitions partitions.
>>>
>>> So I combine them and get number of partitions lower than expected:
>>>
>>> scala> sc.textFile("perf_test1.csv",
>>> minPartitions=128).coalesce(128).getNumPartitions
>>> res14: Int = 126
>>>
>>> Is this expected behaviour? File contains 100000 lines, size of
>>> partitions before and after coalesce:
>>>
>>> scala> sc.textFile("perf_test1.csv",
>>> minPartitions=128).mapPartitions{rows => Iterator(rows.length)}.collect()
>>> res16: Array[Int] = Array(782, 781, 782, 781, 781, 782, 781, 781, 781,
>>> 781, 782, 781, 781, 781, 782, 781, 781, 781, 782, 781, 781, 781, 782, 781,
>>> 781, 781, 782, 781, 781, 782, 781, 781, 781, 781, 782, 781, 781, 781, 782,
>>> 781, 781, 781, 782, 781, 781, 782, 781, 781, 782, 781, 781, 781, 781, 782,
>>> 781, 781, 781, 782, 781, 781, 781, 782, 781, 781, 781, 781, 782, 781, 781,
>>> 782, 781, 781, 782, 781, 781, 781, 781, 782, 781, 781, 781, 782, 781, 781,
>>> 781, 782, 781, 781, 782, 781, 781, 781, 782, 781, 781, 781, 781, 782, 781,
>>> 781, 781, 781, 781, 782, 781, 781, 781, 782, 781, 781, 782, 781, 781, 781,
>>> 781, 782, 781, 781, 781, 781, 782, 781, 781, 782, 781, 781, 781, 781)
>>>
>>> scala> sc.textFile("perf_test1.csv",
>>> minPartitions=128).coalesce(128).mapPartitions{rows =>
>>> Iterator(rows.length)}.collect()
>>> res15: Array[Int] = Array(1563, 781, 781, 781, 782, 781, 781, 781, 781,
>>> 782, 781, 781, 781, 781, 782, 781, 781, 781, 781, 781, 782, 781, 781, 781,
>>> 782, 781, 781, 782, 781, 781, 781, 781, 782, 781, 781, 781, 781, 782, 781,
>>> 781, 782, 781, 781, 781, 781, 1563, 782, 781, 781, 782, 781, 781, 781, 781,
>>> 782, 781, 781, 781, 782, 781, 781, 781, 782, 781, 781, 781, 782, 781, 781,
>>> 781, 782, 781, 781, 782, 781, 781, 781, 781, 782, 781, 781, 781, 782, 781,
>>> 781, 781, 782, 781, 781, 782, 781, 781, 782, 781, 781, 781, 781, 782, 781,
>>> 781, 781, 782, 781, 781, 781, 782, 781, 781, 781, 781, 782, 781, 781, 782,
>>> 781, 781, 782, 781, 781, 781, 781, 782, 781, 781, 781, 782)
>>>
>>> So two partitions are double the size. Is this expected behaviour or is
>>> it some kind of bug?
>>>
>>> Thanks,
>>> Maciej Sokołowski
>>>
>>
>>
>
--
---
Takeshi Yamamuro
Re: Splitting RDD to exact number of partitions
Posted by Maciej Sokołowski <ma...@gmail.com>.
After setting shuffle to true I get expected 128 partitions, but I'm
worried about performance of such solution - especially I see that some
shuffling is done because size of partitions chages:
scala> sc.textFile("hdfs:///proj/dFAB_test/testdata/perf_test1.csv",
minPartitions=128).coalesce(128, true).mapPartitions{rows =>
Iterator(rows.length)}.collect()
res3: Array[Int] = Array(768, 768, 768, 768, 768, 768, 768, 768, 768, 768,
768, 768, 768, 768, 768, 768, 768, 768, 768, 768, 768, 768, 768, 768, 768,
768, 768, 768, 768, 768, 768, 768, 768, 768, 768, 768, 768, 768, 768, 768,
768, 768, 768, 768, 768, 768, 768, 768, 768, 768, 768, 768, 768, 768, 768,
768, 768, 768, 768, 768, 768, 768, 768, 768, 768, 768, 768, 768, 768, 768,
768, 768, 768, 768, 768, 768, 768, 768, 768, 768, 768, 768, 768, 768, 768,
768, 768, 768, 768, 768, 768, 768, 768, 828, 896, 896, 896, 896, 896, 896,
896, 896, 896, 896, 896, 896, 850, 786, 768, 768, 768, 768, 768, 768, 768,
768, 768, 768, 768, 768, 768, 768, 768, 768, 768, 768, 768, 768)
I use spark 1.6.0
On 31 May 2016 at 16:02, Ted Yu <yu...@gmail.com> wrote:
> Value for shuffle is false by default.
>
> Have you tried setting it to true ?
>
> Which Spark release are you using ?
>
> On Tue, May 31, 2016 at 6:13 AM, Maciej Sokołowski <ma...@gmail.com>
> wrote:
>
>> Hello Spark users and developers.
>>
>> I read file and want to ensure that it has exact number of partitions,
>> for example 128.
>>
>> In documentation I found:
>>
>> def textFile(path: String, minPartitions: Int = defaultMinPartitions):
>> RDD[String]
>>
>> But argument here is minimal number of partitions, so I use coalesce to
>> ensure desired number of partitions:
>>
>> def coalesce(numPartitions: Int, shuffle: Boolean = false)(implicit ord:
>> Ordering[T] = null): RDD[T]
>> //Return a new RDD that is reduced into numPartitions partitions.
>>
>> So I combine them and get number of partitions lower than expected:
>>
>> scala> sc.textFile("perf_test1.csv",
>> minPartitions=128).coalesce(128).getNumPartitions
>> res14: Int = 126
>>
>> Is this expected behaviour? File contains 100000 lines, size of
>> partitions before and after coalesce:
>>
>> scala> sc.textFile("perf_test1.csv",
>> minPartitions=128).mapPartitions{rows => Iterator(rows.length)}.collect()
>> res16: Array[Int] = Array(782, 781, 782, 781, 781, 782, 781, 781, 781,
>> 781, 782, 781, 781, 781, 782, 781, 781, 781, 782, 781, 781, 781, 782, 781,
>> 781, 781, 782, 781, 781, 782, 781, 781, 781, 781, 782, 781, 781, 781, 782,
>> 781, 781, 781, 782, 781, 781, 782, 781, 781, 782, 781, 781, 781, 781, 782,
>> 781, 781, 781, 782, 781, 781, 781, 782, 781, 781, 781, 781, 782, 781, 781,
>> 782, 781, 781, 782, 781, 781, 781, 781, 782, 781, 781, 781, 782, 781, 781,
>> 781, 782, 781, 781, 782, 781, 781, 781, 782, 781, 781, 781, 781, 782, 781,
>> 781, 781, 781, 781, 782, 781, 781, 781, 782, 781, 781, 782, 781, 781, 781,
>> 781, 782, 781, 781, 781, 781, 782, 781, 781, 782, 781, 781, 781, 781)
>>
>> scala> sc.textFile("perf_test1.csv",
>> minPartitions=128).coalesce(128).mapPartitions{rows =>
>> Iterator(rows.length)}.collect()
>> res15: Array[Int] = Array(1563, 781, 781, 781, 782, 781, 781, 781, 781,
>> 782, 781, 781, 781, 781, 782, 781, 781, 781, 781, 781, 782, 781, 781, 781,
>> 782, 781, 781, 782, 781, 781, 781, 781, 782, 781, 781, 781, 781, 782, 781,
>> 781, 782, 781, 781, 781, 781, 1563, 782, 781, 781, 782, 781, 781, 781, 781,
>> 782, 781, 781, 781, 782, 781, 781, 781, 782, 781, 781, 781, 782, 781, 781,
>> 781, 782, 781, 781, 782, 781, 781, 781, 781, 782, 781, 781, 781, 782, 781,
>> 781, 781, 782, 781, 781, 782, 781, 781, 782, 781, 781, 781, 781, 782, 781,
>> 781, 781, 782, 781, 781, 781, 782, 781, 781, 781, 781, 782, 781, 781, 782,
>> 781, 781, 782, 781, 781, 781, 781, 782, 781, 781, 781, 782)
>>
>> So two partitions are double the size. Is this expected behaviour or is
>> it some kind of bug?
>>
>> Thanks,
>> Maciej Sokołowski
>>
>
>
Re: Splitting RDD to exact number of partitions
Posted by Ted Yu <yu...@gmail.com>.
Value for shuffle is false by default.
Have you tried setting it to true ?
Which Spark release are you using ?
On Tue, May 31, 2016 at 6:13 AM, Maciej Sokołowski <ma...@gmail.com>
wrote:
> Hello Spark users and developers.
>
> I read file and want to ensure that it has exact number of partitions, for
> example 128.
>
> In documentation I found:
>
> def textFile(path: String, minPartitions: Int = defaultMinPartitions):
> RDD[String]
>
> But argument here is minimal number of partitions, so I use coalesce to
> ensure desired number of partitions:
>
> def coalesce(numPartitions: Int, shuffle: Boolean = false)(implicit ord:
> Ordering[T] = null): RDD[T]
> //Return a new RDD that is reduced into numPartitions partitions.
>
> So I combine them and get number of partitions lower than expected:
>
> scala> sc.textFile("perf_test1.csv",
> minPartitions=128).coalesce(128).getNumPartitions
> res14: Int = 126
>
> Is this expected behaviour? File contains 100000 lines, size of partitions
> before and after coalesce:
>
> scala> sc.textFile("perf_test1.csv", minPartitions=128).mapPartitions{rows
> => Iterator(rows.length)}.collect()
> res16: Array[Int] = Array(782, 781, 782, 781, 781, 782, 781, 781, 781,
> 781, 782, 781, 781, 781, 782, 781, 781, 781, 782, 781, 781, 781, 782, 781,
> 781, 781, 782, 781, 781, 782, 781, 781, 781, 781, 782, 781, 781, 781, 782,
> 781, 781, 781, 782, 781, 781, 782, 781, 781, 782, 781, 781, 781, 781, 782,
> 781, 781, 781, 782, 781, 781, 781, 782, 781, 781, 781, 781, 782, 781, 781,
> 782, 781, 781, 782, 781, 781, 781, 781, 782, 781, 781, 781, 782, 781, 781,
> 781, 782, 781, 781, 782, 781, 781, 781, 782, 781, 781, 781, 781, 782, 781,
> 781, 781, 781, 781, 782, 781, 781, 781, 782, 781, 781, 782, 781, 781, 781,
> 781, 782, 781, 781, 781, 781, 782, 781, 781, 782, 781, 781, 781, 781)
>
> scala> sc.textFile("perf_test1.csv",
> minPartitions=128).coalesce(128).mapPartitions{rows =>
> Iterator(rows.length)}.collect()
> res15: Array[Int] = Array(1563, 781, 781, 781, 782, 781, 781, 781, 781,
> 782, 781, 781, 781, 781, 782, 781, 781, 781, 781, 781, 782, 781, 781, 781,
> 782, 781, 781, 782, 781, 781, 781, 781, 782, 781, 781, 781, 781, 782, 781,
> 781, 782, 781, 781, 781, 781, 1563, 782, 781, 781, 782, 781, 781, 781, 781,
> 782, 781, 781, 781, 782, 781, 781, 781, 782, 781, 781, 781, 782, 781, 781,
> 781, 782, 781, 781, 782, 781, 781, 781, 781, 782, 781, 781, 781, 782, 781,
> 781, 781, 782, 781, 781, 782, 781, 781, 782, 781, 781, 781, 781, 782, 781,
> 781, 781, 782, 781, 781, 781, 782, 781, 781, 781, 781, 782, 781, 781, 782,
> 781, 781, 782, 781, 781, 781, 781, 782, 781, 781, 781, 782)
>
> So two partitions are double the size. Is this expected behaviour or is it
> some kind of bug?
>
> Thanks,
> Maciej Sokołowski
>