[SPARK-19690][SS] Join a streaming DataFrame with a batch DataFrame which has an aggregation may not work by uncleGen · Pull Request #17052 · apache/spark

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.

## What changes were proposed in this pull request? This PR proposes to close stale PRs, mostly the same instances with apache#18017 Closes apache#14085 - [SPARK-16408][SQL] SparkSQL Added file get Exception: is a directory … Closes apache#14239 - [SPARK-16593] [CORE] [WIP] Provide a pre-fetch mechanism to accelerate shuffle stage. Closes apache#14567 - [SPARK-16992][PYSPARK] Python Pep8 formatting and import reorganisation Closes apache#14579 - [SPARK-16921][PYSPARK] RDD/DataFrame persist()/cache() should return Python context managers Closes apache#14601 - [SPARK-13979][Core] Killed executor is re spawned without AWS key… Closes apache#14830 - [SPARK-16992][PYSPARK][DOCS] import sort and autopep8 on Pyspark examples Closes apache#14963 - [SPARK-16992][PYSPARK] Virtualenv for Pylint and pep8 in lint-python Closes apache#15227 - [SPARK-17655][SQL]Remove unused variables declarations and definations in a WholeStageCodeGened stage Closes apache#15240 - [SPARK-17556] [CORE] [SQL] Executor side broadcast for broadcast joins Closes apache#15405 - [SPARK-15917][CORE] Added support for number of executors in Standalone [WIP] Closes apache#16099 - [SPARK-18665][SQL] set statement state to "ERROR" after user cancel job Closes apache#16445 - [SPARK-19043][SQL]Make SparkSQLSessionManager more configurable Closes apache#16618 - [SPARK-14409][ML][WIP] Add RankingEvaluator Closes apache#16766 - [SPARK-19426][SQL] Custom coalesce for Dataset Closes apache#16832 - [SPARK-19490][SQL] ignore case sensitivity when filtering hive partition columns Closes apache#17052 - [SPARK-19690][SS] Join a streaming DataFrame with a batch DataFrame which has an aggregation may not work Closes apache#17267 - [SPARK-19926][PYSPARK] Make pyspark exception more user-friendly Closes apache#17371 - [SPARK-19903][PYSPARK][SS] window operator miss the `watermark` metadata of time column Closes apache#17401 - [SPARK-18364][YARN] Expose metrics for YarnShuffleService Closes apache#17519 - [SPARK-15352][Doc] follow-up: add configuration docs for topology-aware block replication Closes apache#17530 - [SPARK-5158] Access kerberized HDFS from Spark standalone Closes apache#17854 - [SPARK-20564][Deploy] Reduce massive executor failures when executor count is large (>2000) Closes apache#17979 - [SPARK-19320][MESOS][WIP]allow specifying a hard limit on number of gpus required in each spark executor when running on mesos Closes apache#18127 - [SPARK-6628][SQL][Branch-2.1] Fix ClassCastException when executing sql statement 'insert into' on hbase table Closes apache#18236 - [SPARK-21015] Check field name is not null and empty in GenericRowWit… Closes apache#18269 - [SPARK-21056][SQL] Use at most one spark job to list files in InMemoryFileIndex Closes apache#18328 - [SPARK-21121][SQL] Support changing storage level via the spark.sql.inMemoryColumnarStorage.level variable Closes apache#18354 - [SPARK-18016][SQL][CATALYST][BRANCH-2.1] Code Generation: Constant Pool Limit - Class Splitting Closes apache#18383 - [SPARK-21167][SS] Set kafka clientId while fetch messages Closes apache#18414 - [SPARK-21169] [core] Make sure to update application status to RUNNING if executors are accepted and RUNNING after recovery Closes apache#18432 - resolve com.esotericsoftware.kryo.KryoException Closes apache#18490 - [SPARK-21269][Core][WIP] Fix FetchFailedException when enable maxReqSizeShuffleToMem and KryoSerializer Closes apache#18585 - SPARK-21359 Closes apache#18609 - Spark SQL merge small files to big files Update InsertIntoHiveTable.scala Added: Closes apache#18308 - [SPARK-21099][Spark Core] INFO Log Message Using Incorrect Executor I… Closes apache#18599 - [SPARK-21372] spark writes one log file even I set the number of spark_rotate_log to 0 Closes apache#18619 - [SPARK-21397][BUILD]Maven shade plugin adding dependency-reduced-pom.xml to … Closes apache#18667 - Fix the simpleString used in error messages Closes apache#18782 - Branch 2.1 Added: Closes apache#17694 - [SPARK-12717][PYSPARK] Resolving race condition with pyspark broadcasts when using multiple threads Added: Closes apache#16456 - [SPARK-18994] clean up the local directories for application in future by annother thread Closes apache#18683 - [SPARK-21474][CORE] Make number of parallel fetches from a reducer configurable Closes apache#18690 - [SPARK-21334][CORE] Add metrics reporting service to External Shuffle Server Added: Closes apache#18827 - Merge pull request 1 from apache/master ## How was this patch tested? N/A Author: hyukjinkwon <gurwls223@gmail.com> Closes apache#18780 from HyukjinKwon/close-prs.

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

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

Uh oh!

SparkQA commented Mar 21, 2017

Uh oh!

zsxwing commented Feb 24, 2017 •

edited

Loading

uncleGen Feb 28, 2017 •

edited

Loading