[SPARK-19690][SS] Join a streaming DataFrame with a batch DataFrame which has an aggregation may not work #17052

uncleGen · 2017-02-24T06:41:39Z

What changes were proposed in this pull request?

StatefulAggregationStrategy should check logicplan is streaming or not

Test code:

case class Record(key: Int, value: String)
val df = spark.createDataFrame((1 to 100).map(i => Record(i, s"value_$i"))).groupBy("value").count
val lines = spark.readStream.format("socket").option("host", "localhost").option("port", "9999").load 
val words = lines.as[String].flatMap(_.split(" ")) 
val result = words.join(df, "value")

before pr:

== Physical Plan ==
*Project [value#13, count#19L]
+- *BroadcastHashJoin [value#13], [value#1], Inner, BuildRight
   :- *Filter isnotnull(value#13)
   :  +- *SerializeFromObject [staticinvoke(class org.apache.spark.unsafe.types.UTF8String, StringType, fromString, input[0, java.lang.String, true], true) AS value#13]
   :     +- MapPartitions <function1>, obj#12: java.lang.String
   :        +- DeserializeToObject value#5.toString, obj#11: java.lang.String
   :           +- StreamingRelation textSocket, [value#5]
   +- BroadcastExchange HashedRelationBroadcastMode(List(input[0, string, true]))
      +- *HashAggregate(keys=[value#1], functions=[count(1)])
         +- StateStoreSave [value#1], OperatorStateId(<unknown>,0,0), Append, 0
            +- *HashAggregate(keys=[value#1], functions=[merge_count(1)])
               +- StateStoreRestore [value#1], OperatorStateId(<unknown>,0,0)
                  +- *HashAggregate(keys=[value#1], functions=[merge_count(1)])
                     +- Exchange hashpartitioning(value#1, 200)
                        +- *HashAggregate(keys=[value#1], functions=[partial_count(1)])
                           +- *Project [value#1]
                              +- *Filter isnotnull(value#1)
                                 +- LocalTableScan [key#0, value#1]

after pr:

== Physical Plan ==
*Project [value#13, count#19L]
+- *BroadcastHashJoin [value#13], [value#1], Inner, BuildRight
   :- *Filter isnotnull(value#13)
   :  +- *SerializeFromObject [staticinvoke(class org.apache.spark.unsafe.types.UTF8String, StringType, fromString, input[0, java.lang.String, true], true) AS value#13]
   :     +- MapPartitions <function1>, obj#12: java.lang.String
   :        +- DeserializeToObject value#5.toString, obj#11: java.lang.String
   :           +- StreamingRelation textSocket, [value#5]
   +- BroadcastExchange HashedRelationBroadcastMode(List(input[0, string, true]))
      +- *HashAggregate(keys=[value#1], functions=[count(1)])
         +- Exchange hashpartitioning(value#1, 200)
            +- *HashAggregate(keys=[value#1], functions=[partial_count(1)])
               +- *Project [value#1]
                  +- *Filter isnotnull(value#1)
                     +- LocalTableScan [key#0, value#1]

How was this patch tested?

add new unit test.

SparkQA · 2017-02-24T07:12:44Z

Test build #73407 has started for PR 17052 at commit e8a24e1.

SparkQA · 2017-02-24T08:02:08Z

Test build #73403 has finished for PR 17052 at commit 9eb57b7.

This patch fails Spark unit tests.
This patch merges cleanly.
This patch adds no public classes.

SparkQA · 2017-02-24T10:12:59Z

Test build #73412 has finished for PR 17052 at commit 9c15fcb.

This patch passes all tests.
This patch merges cleanly.
This patch adds no public classes.

uncleGen · 2017-02-24T10:20:44Z

cc @zsxwing

uncleGen · 2017-02-24T11:13:41Z

...alyst/src/main/scala/org/apache/spark/sql/catalyst/plans/logical/basicLogicalOperators.scala

stateful indicates if the aggregate is base on streaming or batch, resolved by ResolveStatefulAggregate rule

uncleGen · 2017-02-24T11:14:32Z

sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/rules.scala

resolve one aggregate, determine statefule or not.

zsxwing · 2017-02-24T18:50:15Z

Thanks for doing this. I'm wondering if you can fix isStreaming instead. We added it to be able to distinguish batch and streaming dataframes. However, it doesn't work for batch DFs in a streaming query. My initial thought is making sure all streaming Source's getBatch should return a DataFrame whose isStreaming must return true. Could you try to resolve this issue in this direction?

uncleGen · 2017-02-25T03:00:04Z

@zsxwing got it

uncleGen · 2017-02-27T05:06:56Z

retest this please.

SparkQA · 2017-02-27T18:09:51Z

Test build #73507 has finished for PR 17052 at commit 38e3a14.

This patch fails Spark unit tests.
This patch merges cleanly.
This patch adds no public classes.

uncleGen · 2017-02-28T01:07:20Z

working on unit test failure

uncleGen · 2017-02-28T03:54:56Z

retest this please.

SparkQA · 2017-02-28T05:14:34Z

Test build #73559 has finished for PR 17052 at commit 59f4272.

This patch fails Spark unit tests.
This patch merges cleanly.
This patch adds no public classes.

SparkQA · 2017-02-28T05:51:54Z

Test build #73558 has finished for PR 17052 at commit c87651a.

This patch passes all tests.
This patch merges cleanly.
This patch adds no public classes.

SparkQA · 2017-02-28T06:27:30Z

Test build #73571 has started for PR 17052 at commit 9ffbad2.

uncleGen · 2017-02-28T06:26:49Z

sql/core/src/test/scala/org/apache/spark/sql/streaming/FileStreamSourceSuite.scala

-        SQLConf.FILE_SOURCE_LOG_CLEANUP_DELAY.key -> "1"
+        SQLConf.FILE_SOURCE_LOG_CLEANUP_DELAY.key -> "1",
+        SQLConf.UNSUPPORTED_OPERATION_CHECK_ENABLED.key -> "false"
      ) {


Close the "UNSUPPORTED_OPERATION_CHECK_ENABLED", as Source.getBatch returns DF whose isStreaming is true.

uncleGen · 2017-02-28T06:31:05Z

sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/Optimizer.scala

+      } else {
+        LocalRelation(projectList.map(_.toAttribute), data.map(projection))
+      }
  }


In a streaming query, we will transfrom stream source to a batch LocalRelation whose isStreaming is true, so we should keep new LocalRelation's isStreaming is true in this rule.

uncleGen · 2017-02-28T06:31:57Z

sql/core/src/main/scala/org/apache/spark/sql/execution/SparkStrategies.scala

+      case agg @ PhysicalAggregation(
+        namedGroupingExpressions, aggregateExpressions, rewrittenResultExpressions, child)
+        if agg.isStreaming =>



Apply this strategy only if the logical plan is streaming.

uncleGen · 2017-02-28T08:12:24Z

retest this please.

SparkQA · 2017-02-28T10:20:08Z

Test build #73576 has finished for PR 17052 at commit 9ffbad2.

This patch passes all tests.
This patch merges cleanly.
This patch adds no public classes.

uncleGen · 2017-03-02T01:20:24Z

\cc @zsxwing

uncleGen · 2017-03-03T02:23:24Z

ping @zsxwing

zsxwing · 2017-03-04T06:26:58Z

sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/plans/logical/LogicalPlan.scala


  private var _analyzed: Boolean = false

+  private var _incremental: Boolean = false


Adding it here will break sameResult, equals and other methods. Could you add a new parameter to the constructor of LogicalRelation and LogicalRDD instead?

uncleGen · 2017-03-07T12:06:44Z

sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/Source.scala

+      case localRelation @ LocalRelation(_, _, false) =>
+        localRelation.dataFromStreaming = true
+        localRelation
+    }


add a new parameter dataFromStreaming to the constructor of LogicalRelation, LogicalRDD and LocalRelation. dataFromStreaming indicate if this relation comes from a streaming source. In a streaming query, stream relation will be cut into a series of batch relations.

SparkQA · 2017-03-07T13:19:58Z

Test build #74099 has finished for PR 17052 at commit 67847e5.

This patch fails Spark unit tests.
This patch merges cleanly.
This patch adds the following public classes (experimental):
case class LocalRelation(

SparkQA · 2017-03-07T13:24:16Z

Test build #74100 has finished for PR 17052 at commit 3fc31d8.

This patch fails Spark unit tests.
This patch merges cleanly.
This patch adds no public classes.

SparkQA · 2017-03-07T17:22:03Z

Test build #74109 has finished for PR 17052 at commit 186be58.

This patch passes all tests.
This patch merges cleanly.
This patch adds no public classes.

SparkQA · 2017-03-08T14:43:59Z

Test build #74206 has finished for PR 17052 at commit ff38db7.

This patch passes all tests.
This patch merges cleanly.
This patch adds the following public classes (experimental):
public static class LongWrapper
public static class IntWrapper
trait WatermarkSupport extends UnaryExecNode

SparkQA · 2017-03-20T04:30:58Z

Test build #74844 has finished for PR 17052 at commit a1f9327.

This patch fails Spark unit tests.
This patch merges cleanly.
This patch adds no public classes.

uncleGen · 2017-03-20T04:36:32Z

retest this please.

SparkQA · 2017-03-20T06:58:26Z

Test build #74852 has finished for PR 17052 at commit a1f9327.

This patch fails SparkR unit tests.
This patch merges cleanly.
This patch adds no public classes.

uncleGen · 2017-03-21T09:13:43Z

retest this please.

SparkQA · 2017-03-21T11:30:36Z

Test build #74968 has finished for PR 17052 at commit a1f9327.

This patch passes all tests.
This patch merges cleanly.
This patch adds no public classes.

srowen · 2017-03-28T21:30:32Z

This is marked as "Critical" for 2.1.1, but I'm not clear it's a regression or that urgent?

HyukjinKwon · 2017-05-11T14:36:16Z

@uncleGen is this still active?

uncleGen · 2017-05-12T05:10:41Z

@HyukjinKwon Sorry! Busy for this period of time. Let me resolve this conflict.

HyukjinKwon · 2017-05-12T05:16:34Z

Yea, I just wanted to check if it is in progress in any way. Thanks for your input.

SparkQA · 2017-06-29T12:21:59Z

Test build #78895 has finished for PR 17052 at commit a1f9327.

This patch passes all tests.
This patch does not merge cleanly.
This patch adds no public classes.

uncleGen commented Feb 24, 2017

View reviewed changes

uncleGen force-pushed the SPARK-19690 branch from 9c15fcb to 38e3a14 Compare February 27, 2017 05:05

uncleGen force-pushed the SPARK-19690 branch 2 times, most recently from c87651a to 59f4272 Compare February 28, 2017 03:53

uncleGen commented Feb 28, 2017

View reviewed changes

zsxwing reviewed Mar 4, 2017

View reviewed changes

uncleGen added 3 commits March 7, 2017 19:54

Streaming Source getBatch should return a streaming DataFrame

0363853

fix test failure

9caa2d5

address comments from @zsxwing

67847e5

uncleGen force-pushed the SPARK-19690 branch 2 times, most recently from c87651a to 67847e5 Compare March 7, 2017 11:56

update doc

3fc31d8

uncleGen commented Mar 7, 2017

View reviewed changes

fix test failure

186be58

Merge branch 'master' into SPARK-19690

ff38db7

Merge branch 'master' into SPARK-19690

a1f9327

HyukjinKwon mentioned this pull request Jul 31, 2017

[INFRA] Close stale PRs #18780

Closed

asfgit closed this in 3a45c7f Aug 5, 2017


		private var _analyzed: Boolean = false

		private var _incremental: Boolean = false

[SPARK-19690][SS] Join a streaming DataFrame with a batch DataFrame which has an aggregation may not work #17052

[SPARK-19690][SS] Join a streaming DataFrame with a batch DataFrame which has an aggregation may not work #17052

Uh oh!

Conversation

uncleGen commented Feb 24, 2017

What changes were proposed in this pull request?

How was this patch tested?

Uh oh!

SparkQA commented Feb 24, 2017

Uh oh!

SparkQA commented Feb 24, 2017

Uh oh!

SparkQA commented Feb 24, 2017

Uh oh!

uncleGen commented Feb 24, 2017

Uh oh!

uncleGen Feb 24, 2017

Choose a reason for hiding this comment

Uh oh!

uncleGen Feb 24, 2017

Choose a reason for hiding this comment

Uh oh!

zsxwing commented Feb 24, 2017 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

uncleGen commented Feb 25, 2017

Uh oh!

uncleGen commented Feb 27, 2017

Uh oh!

SparkQA commented Feb 27, 2017

Uh oh!

uncleGen commented Feb 28, 2017

Uh oh!

uncleGen commented Feb 28, 2017

Uh oh!

SparkQA commented Feb 28, 2017

Uh oh!

SparkQA commented Feb 28, 2017

Uh oh!

SparkQA commented Feb 28, 2017

Uh oh!

uncleGen Feb 28, 2017

Choose a reason for hiding this comment

Uh oh!

uncleGen Feb 28, 2017

Choose a reason for hiding this comment

Uh oh!

uncleGen Feb 28, 2017 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

uncleGen commented Feb 28, 2017

Uh oh!

SparkQA commented Feb 28, 2017

Uh oh!

uncleGen commented Mar 2, 2017

Uh oh!

uncleGen commented Mar 3, 2017

Uh oh!

zsxwing Mar 4, 2017

Choose a reason for hiding this comment

Uh oh!

uncleGen Mar 7, 2017

Choose a reason for hiding this comment

Uh oh!

SparkQA commented Mar 7, 2017

Uh oh!

SparkQA commented Mar 7, 2017

Uh oh!

SparkQA commented Mar 7, 2017

Uh oh!

SparkQA commented Mar 8, 2017

Uh oh!

SparkQA commented Mar 20, 2017

Uh oh!

uncleGen commented Mar 20, 2017

Uh oh!

SparkQA commented Mar 20, 2017

Uh oh!

uncleGen commented Mar 21, 2017

zsxwing commented Feb 24, 2017 •

edited

Loading

uncleGen Feb 28, 2017 •

edited

Loading