I understand the actual dataframe is different, but the underlying partitions are not (hence the importance of mark's response). The code you suggested would not work as allDF and x would have different schema's (x is the original and allDF becomes the grouped). I can do something like this: var totalTime: Long = 0 var allDF: DataFrame = null for { x <- dataframes } { val timeLen = time { val grouped = x.groupBy("cat1", "cat2").agg(sum($"valToAdd").alias("v")) allDF = if (allDF == null) grouped else { allDF.union(grouped).groupBy("cat1", "cat2").agg(sum($"v").alias("v")) } val grouped2 = allDF.groupBy("cat1").agg(sum($"v"), count($"cat2")) grouped2.show() } totalTime += timeLen println(s"Took $timeLen miliseconds") } println(s"Overall time was $totalTime miliseconds") }
and this indeed improves performance (I actually had a couple more tries) but: 1. This still gives crappy performance (for 167 slices I get a throughput which is 10 times lower than batch after doing some tuning including caching and coalescing) 2. This works because the aggregation here is sum and we don't forget. For more general aggregations we would have to join them together (can't do it for count distinct for example) and we will need to "forget" frames when moving out of the window (we can subtract a sum but not a max). The best solution I found so far (performance wise) was to write a custom UDAF which does the window internally. This was still 8 times lower throughput than batch and required a lot of coding and is not a general solution. I am looking for an approach to improve the performance even more (preferably to either be on par with batch or a relatively low factor which remains constant when the number of slices rise) and including the option to "forget" frames. Assaf. From: Liang-Chi Hsieh [via Apache Spark Developers List] [mailto:ml-node+s1001551n20371...@n3.nabble.com] Sent: Wednesday, December 28, 2016 3:59 AM To: Mendelson, Assaf Subject: RE: Shuffle intermidiate results not being cached Hi, Every iteration the data you run aggregation on it is different. As I showed in previous reply: 1st iteration: aggregation(x1 union x2) 2nd iteration: aggregation(x3 union (x1 union x2)) 3rd iteration: aggregation(x4 union(x3 union (x1 union x2))) In 1st you run aggregation on the data of x1 and x2. In 2nd the data is x1, x2 and x3. Even you work on the same RDD, you won't see reuse of the shuffle data because the shuffle data is different. In your second example, I think the way to reduce the computation is like: var totalTime: Long = 0 var allDF: org.apache.spark.sql.DataFrame = null for { x <- dataframes } { val timeLen = time { allDF = if (allDF == null) x else allDF.union(x) // Union previous aggregation summary with new dataframe in this window val grouped = allDF.groupBy("cat1", "cat2").agg(sum($"valToAdd").alias("v")) val grouped2 = grouped.groupBy("cat1").agg(sum($"v"), count($"cat2")) grouped2.show() allDF = grouped // Replace the union of data with aggregated summary } totalTime += timeLen println(s"Took $timeLen miliseconds") } println(s"Total time was $totalTime miliseconds") You don't need to recompute the aggregation of previous dataframes in each iteration. You just need to get the summary and union it with new dataframe to compute the newer aggregation summary in next iteration. It is more similar to streaming case, I don't think you can/should recompute all the data since the beginning of a stream. assaf.mendelson wrote The reason I thought some operations would be reused is the fact that spark automatically caches shuffle data which means the partial aggregation for pervious dataframes would be saved. Unfortunatly, as Mark Hamstra explained this is not the case because this is considered a new RDD and therefore the previous data is lost. I am still wondering if there is any way to do high performance streaming of SQL. Basically this is not far from what DStream would do assuming we convert a sliding window (e.g. 24 hours every 15 minutes) as we would be doing a foreachRDD which would do the joining behind the scenes. The problem is that any attempt to do a streaming like this results in performance which is hundreds of times slower than batch. Is there a correct way to do such an aggregation on streaming data (using dataframes rather than RDD operations). Assaf. From: Liang-Chi Hsieh [via Apache Spark Developers List] [mailto:[hidden email]</user/SendEmail.jtp?type=node&node=20371&i=0>] Sent: Monday, December 26, 2016 5:42 PM To: Mendelson, Assaf Subject: Re: Shuffle intermidiate results not being cached Hi, Let me quote your example codes: var totalTime: Long = 0 var allDF: org.apache.spark.sql.DataFrame = null for { x <- dataframes } { val timeLen = time { allDF = if (allDF == null) x else allDF.union(x) val grouped = allDF.groupBy("cat1", "cat2").agg(sum($"valToAdd").alias("v")) val grouped2 = grouped.groupBy("cat1").agg(sum($"v"), count($"cat2")) grouped2.show() } totalTime += timeLen println(s"Took $timeLen miliseconds") } println(s"Total time was $totalTime miliseconds") Basically what you do is to union some dataframes for each iteration, and do aggregation on this union data. I don't see any reused operations. 1st iteration: aggregation(x1 union x2) 2nd iteration: aggregation(x3 union (x1 union x2)) 3rd iteration: aggregation(x4 union(x3 union (x1 union x2))) ... Your first example just does two aggregation operations. But your second example like above does this aggregation operations for each iteration. So the time of second example grows as the iteration increases. Liang-Chi Hsieh | @viirya Spark Technology Center http://www.spark.tc/ ________________________________ If you reply to this email, your message will be added to the discussion below: http://apache-spark-developers-list.1001551.n3.nabble.com/Shuffle-intermidiate-results-not-being-cached-tp20358p20361.html To start a new topic under Apache Spark Developers List, email [hidden email]</user/SendEmail.jtp?type=node&node=20371&i=1><mailto:[hidden email]</user/SendEmail.jtp?type=node&node=20371&i=2>> To unsubscribe from Apache Spark Developers List, click here<http://apache-spark-developers-list.1001551.n3.nabble.com/template/NamlServlet.jtp?macro=unsubscribe_by_code&node=1&code=YXNzYWYubWVuZGVsc29uQHJzYS5jb218MXwtMTI4OTkxNTg1Mg==><http://apache-spark-developers-list.1001551.n3.nabble.com/template/NamlServlet.jtp?macro=unsubscribe_by_code&node=1&code=YXNzYWYubWVuZGVsc29uQHJzYS5jb218MXwtMTI4OTkxNTg1Mg==%3e>. NAML<http://apache-spark-developers-list.1001551.n3.nabble.com/template/NamlServlet.jtp?macro=macro_viewer&id=instant_html%21nabble%3Aemail.naml&base=nabble.naml.namespaces.BasicNamespace-nabble.view.web.template.NabbleNamespace-nabble.view.web.template.NodeNamespace&breadcrumbs=notify_subscribers%21nabble%3Aemail.naml-instant_emails%21nabble%3Aemail.naml-send_instant_email%21nabble%3Aemail.naml><http://apache-spark-developers-list.1001551.n3.nabble.com/template/NamlServlet.jtp?macro=macro_viewer&id=instant_html%21nabble%3Aemail.naml&base=nabble.naml.namespaces.BasicNamespace-nabble.view.web.template.NabbleNamespace-nabble.view.web.template.NodeNamespace&breadcrumbs=notify_subscribers%21nabble%3Aemail.naml-instant_emails%21nabble%3Aemail.naml-send_instant_email%21nabble%3Aemail.naml%3e> Liang-Chi Hsieh | @viirya Spark Technology Center http://www.spark.tc/ ________________________________ If you reply to this email, your message will be added to the discussion below: http://apache-spark-developers-list.1001551.n3.nabble.com/Shuffle-intermidiate-results-not-being-cached-tp20358p20371.html To start a new topic under Apache Spark Developers List, email ml-node+s1001551n1...@n3.nabble.com<mailto:ml-node+s1001551n1...@n3.nabble.com> To unsubscribe from Apache Spark Developers List, click here<http://apache-spark-developers-list.1001551.n3.nabble.com/template/NamlServlet.jtp?macro=unsubscribe_by_code&node=1&code=YXNzYWYubWVuZGVsc29uQHJzYS5jb218MXwtMTI4OTkxNTg1Mg==>. NAML<http://apache-spark-developers-list.1001551.n3.nabble.com/template/NamlServlet.jtp?macro=macro_viewer&id=instant_html%21nabble%3Aemail.naml&base=nabble.naml.namespaces.BasicNamespace-nabble.view.web.template.NabbleNamespace-nabble.view.web.template.NodeNamespace&breadcrumbs=notify_subscribers%21nabble%3Aemail.naml-instant_emails%21nabble%3Aemail.naml-send_instant_email%21nabble%3Aemail.naml> -- View this message in context: http://apache-spark-developers-list.1001551.n3.nabble.com/Shuffle-intermidiate-results-not-being-cached-tp20358p20377.html Sent from the Apache Spark Developers List mailing list archive at Nabble.com.