[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

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

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

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

- 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

+ } 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

+ 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


 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

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

uncleGen commented Feb 24, 2017
View reviewed changes

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

uncleGen force-pushed the SPARK-19690 branch 2 times, most recently from c87651a to 59f4272Compare 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 67847e5Compare 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 3a45c7fAug 5, 2017


		private var _analyzed: Boolean = false

		private var _incremental: Boolean = false

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!

uncleGenFeb 24, 2017

Choose a reason for hiding this comment

Uh oh!

uncleGenFeb 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!

uncleGenFeb 28, 2017

Choose a reason for hiding this comment

Uh oh!

uncleGenFeb 28, 2017

Choose a reason for hiding this comment

Uh oh!

uncleGenFeb 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!

zsxwingMar 4, 2017

Choose a reason for hiding this comment

Uh oh!

uncleGenMar 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

zsxwing commented Feb 24, 2017•
edited
Loading

uncleGenFeb 28, 2017•
edited
Loading