You are viewing a plain text version of this content. The canonical link for it is here.
Posted to user@spark.apache.org by onmstester onmstester <on...@zoho.com.INVALID> on 2018/11/21 04:34:09 UTC
spark-sql force parallel union
I'm using Spark-Sql to query Cassandra tables. In Cassandra, i've partitioned my data with time bucket and one id, so based on queries i need to union multiple partitions with spark-sql and do the aggregations/group-by on union-result, something like this: for(all cassandra partitions){
DataSet<Row> currentPartition = sqlContext.sql(....);
unionResult = unionResult.union(currentPartition);
}
Increasing input (number of loaded partitions), increases response time more than linearly because unions would be done sequentialy. Because there is no harm in doing unions in parallel, and i dont know how to force spark to do them in parallel, Right now i'm using a ThreadPool to Asyncronosly load all partitions in my application (which may cause OOM), and somehow do the sort or simple group by in java (Which make me think why even i'm using spark at all?) The short question is: How to force spark-sql to load cassandra partitions in parallel while doing union on them? Also I don't want too many tasks in spark, with my Home-Made Async solution, i use coalesece(1) so one task is so fast (only wait time on casandra). Sent using Zoho Mail
Re: Re: spark-sql force parallel union
Posted by Alessandro Solimando <al...@gmail.com>.
Hello,
maybe I am overlooking the problem but what I would go for something
similar:
def unionDFs(dfs: List[DataFrame]): DataFrame = {
dfs.drop(0).foldRight(dfs.apply(0))((df1: DataFrame, df2: DataFrame) =>
df1 union df2)
}
(Would be better to keep dfs as-is and you use an empty DF with the correct
schema).
This should create the sought DAG plan, I cannot give it a run at the
moment to confirm.
At this point you need no views and you will benefit from the parallelism.
What do you think?
Best regards,
Alessandro
On Wed, 21 Nov 2018 at 08:19, onmstester onmstester
<on...@zoho.com.invalid> wrote:
> Thanks Kathleen,
>
> 1. So if i've got 4 df's and i want "dfv1 union dfv2 union dfv3 union
> dfv4", would it first compute "dfv1 union dfv2" and "dfv3 union dfv4"
> independently and simultaneously? then union their results?
> 2. Its going to be hundreds of partitions to union, creating a temp view
> for each of them might be slow?
>
> Sent using Zoho Mail <https://www.zoho.com/mail/>
>
>
> ============ Forwarded message ============
> From : kathleen li <ka...@gmail.com>
> To : <on...@zoho.com.invalid>
> Cc : <us...@spark.apache.org>
> Date : Wed, 21 Nov 2018 10:16:21 +0330
> Subject : Re: spark-sql force parallel union
> ============ Forwarded message ============
>
> you might first write the code to construct query statement with "union
> all" like below:
>
> scala> val query="select * from dfv1 union all select * from dfv2 union
> all select * from dfv3"
> query: String = select * from dfv1 union all select * from dfv2 union all
> select * from dfv3
>
> then write loop to register each partition to a view like below:
> for (i <- 1 to 3){
> df.createOrReplaceTempView("dfv"+i)
> }
>
> scala> spark.sql(query).explain
> == Physical Plan ==
> Union
> :- LocalTableScan [_1#0, _2#1, _3#2]
> :- LocalTableScan [_1#0, _2#1, _3#2]
> +- LocalTableScan [_1#0, _2#1, _3#2]
>
>
> You can use " roll up" or "group set" for multiple dimension to replace
> "union" or "union all"
>
> On Tue, Nov 20, 2018 at 8:34 PM onmstester onmstester <
> onmstester@zoho.com.invalid> wrote:
>
>
> I'm using Spark-Sql to query Cassandra tables. In Cassandra, i've
> partitioned my data with time bucket and one id, so based on queries i need
> to union multiple partitions with spark-sql and do the
> aggregations/group-by on union-result, something like this:
>
> for(all cassandra partitions){
> DataSet<Row> currentPartition = sqlContext.sql(....);
> unionResult = unionResult.union(currentPartition);
> }
>
> Increasing input (number of loaded partitions), increases response time
> more than linearly because unions would be done sequentialy.
>
> Because there is no harm in doing unions in parallel, and i dont know how
> to force spark to do them in parallel, Right now i'm using a ThreadPool to
> Asyncronosly load all partitions in my application (which may cause OOM),
> and somehow do the sort or simple group by in java (Which make me think why
> even i'm using spark at all?)
>
> The short question is: How to force spark-sql to load cassandra partitions
> in parallel while doing union on them? Also I don't want too many tasks in
> spark, with my Home-Made Async solution, i use coalesece(1) so one task is
> so fast (only wait time on casandra).
>
> Sent using Zoho Mail <https://www.zoho.com/mail/>
>
>
>
>
Fwd: Re: spark-sql force parallel union
Posted by onmstester onmstester <on...@zoho.com.INVALID>.
Thanks Kathleen, 1. So if i've got 4 df's and i want "dfv1 union dfv2 union dfv3 union dfv4", would it first compute "dfv1 union dfv2" and "dfv3 union dfv4" independently and simultaneously? then union their results? 2. Its going to be hundreds of partitions to union, creating a temp view for each of them might be slow? Sent using Zoho Mail ============ Forwarded message ============ From : kathleen li <ka...@gmail.com> To : <on...@zoho.com.invalid> Cc : <us...@spark.apache.org> Date : Wed, 21 Nov 2018 10:16:21 +0330 Subject : Re: spark-sql force parallel union ============ Forwarded message ============ you might first write the code to construct query statement with "union all" like below: scala> val query="select * from dfv1 union all select * from dfv2 union all select * from dfv3" query: String = select * from dfv1 union all select * from dfv2 union all select * from dfv3 then write loop to register each partition to a view like below: for (i <- 1 to 3){ df.createOrReplaceTempView("dfv"+i) } scala> spark.sql(query).explain == Physical Plan == Union :- LocalTableScan [_1#0, _2#1, _3#2] :- LocalTableScan [_1#0, _2#1, _3#2] +- LocalTableScan [_1#0, _2#1, _3#2] You can use " roll up" or "group set" for multiple dimension to replace "union" or "union all" On Tue, Nov 20, 2018 at 8:34 PM onmstester onmstester <on...@zoho.com.invalid> wrote: I'm using Spark-Sql to query Cassandra tables. In Cassandra, i've partitioned my data with time bucket and one id, so based on queries i need to union multiple partitions with spark-sql and do the aggregations/group-by on union-result, something like this: for(all cassandra partitions){
DataSet<Row> currentPartition = sqlContext.sql(....);
unionResult = unionResult.union(currentPartition);
}
Increasing input (number of loaded partitions), increases response time more than linearly because unions would be done sequentialy. Because there is no harm in doing unions in parallel, and i dont know how to force spark to do them in parallel, Right now i'm using a ThreadPool to Asyncronosly load all partitions in my application (which may cause OOM), and somehow do the sort or simple group by in java (Which make me think why even i'm using spark at all?) The short question is: How to force spark-sql to load cassandra partitions in parallel while doing union on them? Also I don't want too many tasks in spark, with my Home-Made Async solution, i use coalesece(1) so one task is so fast (only wait time on casandra). Sent using Zoho Mail
Re: spark-sql force parallel union
Posted by kathleen li <ka...@gmail.com>.
you might first write the code to construct query statement with "union
all" like below:
scala> val query="select * from dfv1 union all select * from dfv2 union all
select * from dfv3"
query: String = select * from dfv1 union all select * from dfv2 union all
select * from dfv3
then write loop to register each partition to a view like below:
for (i <- 1 to 3){
df.createOrReplaceTempView("dfv"+i)
}
scala> spark.sql(query).explain
== Physical Plan ==
Union
:- LocalTableScan [_1#0, _2#1, _3#2]
:- LocalTableScan [_1#0, _2#1, _3#2]
+- LocalTableScan [_1#0, _2#1, _3#2]
You can use " roll up" or "group set" for multiple dimension to replace
"union" or "union all"
On Tue, Nov 20, 2018 at 8:34 PM onmstester onmstester
<on...@zoho.com.invalid> wrote:
> I'm using Spark-Sql to query Cassandra tables. In Cassandra, i've
> partitioned my data with time bucket and one id, so based on queries i need
> to union multiple partitions with spark-sql and do the
> aggregations/group-by on union-result, something like this:
>
> for(all cassandra partitions){
> DataSet<Row> currentPartition = sqlContext.sql(....);
> unionResult = unionResult.union(currentPartition);
> }
>
> Increasing input (number of loaded partitions), increases response time
> more than linearly because unions would be done sequentialy.
>
> Because there is no harm in doing unions in parallel, and i dont know how
> to force spark to do them in parallel, Right now i'm using a ThreadPool to
> Asyncronosly load all partitions in my application (which may cause OOM),
> and somehow do the sort or simple group by in java (Which make me think why
> even i'm using spark at all?)
>
> The short question is: How to force spark-sql to load cassandra partitions
> in parallel while doing union on them? Also I don't want too many tasks in
> spark, with my Home-Made Async solution, i use coalesece(1) so one task is
> so fast (only wait time on casandra).
>
> Sent using Zoho Mail <https://www.zoho.com/mail/>
>
>
>