You are viewing a plain text version of this content. The canonical link for it is here.
Posted to reviews@spark.apache.org by "jingz-db (via GitHub)" <gi...@apache.org> on 2024/09/16 20:11:52 UTC
[PR] [SS] Add Support for Chaining of Operators in transformWithStateInPandas API [spark]
jingz-db opened a new pull request, #48124:
URL: https://github.com/apache/spark/pull/48124
<!--
Thanks for sending a pull request! Here are some tips for you:
1. If this is your first time, please read our contributor guidelines: https://spark.apache.org/contributing.html
2. Ensure you have added or run the appropriate tests for your PR: https://spark.apache.org/developer-tools.html
3. If the PR is unfinished, add '[WIP]' in your PR title, e.g., '[WIP][SPARK-XXXX] Your PR title ...'.
4. Be sure to keep the PR description updated to reflect all changes.
5. Please write your PR title to summarize what this PR proposes.
6. If possible, provide a concise example to reproduce the issue for a faster review.
7. If you want to add a new configuration, please read the guideline first for naming configurations in
'core/src/main/scala/org/apache/spark/internal/config/ConfigEntry.scala'.
8. If you want to add or modify an error type or message, please read the guideline first in
'common/utils/src/main/resources/error/README.md'.
-->
### What changes were proposed in this pull request?
<!--
Please clarify what changes you are proposing. The purpose of this section is to outline the changes and how this PR fixes the issue.
If possible, please consider writing useful notes for better and faster reviews in your PR. See the examples below.
1. If you refactor some codes with changing classes, showing the class hierarchy will help reviewers.
2. If you fix some SQL features, you can provide some references of other DBMSes.
3. If there is design documentation, please add the link.
4. If there is a discussion in the mailing list, please add the link.
-->
### Why are the changes needed?
<!--
Please clarify why the changes are needed. For instance,
1. If you propose a new API, clarify the use case for a new API.
2. If you fix a bug, you can clarify why it is a bug.
-->
### Does this PR introduce _any_ user-facing change?
<!--
Note that it means *any* user-facing change including all aspects such as the documentation fix.
If yes, please clarify the previous behavior and the change this PR proposes - provide the console output, description and/or an example to show the behavior difference if possible.
If possible, please also clarify if this is a user-facing change compared to the released Spark versions or within the unreleased branches such as master.
If no, write 'No'.
-->
### How was this patch tested?
<!--
If tests were added, say they were added here. Please make sure to add some test cases that check the changes thoroughly including negative and positive cases if possible.
If it was tested in a way different from regular unit tests, please clarify how you tested step by step, ideally copy and paste-able, so that other reviewers can test and check, and descendants can verify in the future.
If tests were not added, please describe why they were not added and/or why it was difficult to add.
If benchmark tests were added, please run the benchmarks in GitHub Actions for the consistent environment, and the instructions could accord to: https://spark.apache.org/developer-tools.html#github-workflow-benchmarks.
-->
### Was this patch authored or co-authored using generative AI tooling?
<!--
If generative AI tooling has been used in the process of authoring this patch, please include the
phrase: 'Generated-by: ' followed by the name of the tool and its version.
If no, write 'No'.
Please refer to the [ASF Generative Tooling Guidance](https://www.apache.org/legal/generative-tooling.html) for details.
-->
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-49676][SS][PYTHON] Add Support for Chaining of Operators in transformWithStateInPandas API [spark]
Posted by "HeartSaVioR (via GitHub)" <gi...@apache.org>.
HeartSaVioR commented on code in PR #48124:
URL: https://github.com/apache/spark/pull/48124#discussion_r1859462336
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/SparkStrategies.scala:
##########
@@ -966,6 +966,9 @@ abstract class SparkStrategies extends QueryPlanner[SparkPlan] {
case _: FlatMapGroupsInPandasWithState =>
// TODO(SPARK-40443): support applyInPandasWithState in batch query
throw new SparkUnsupportedOperationException("_LEGACY_ERROR_TEMP_3176")
+ case _: TransformWithStateInPandas =>
+ // TODO(SPARK-50428): support TransformWithStateInPandas in batch query
+ throw new SparkUnsupportedOperationException("_LEGACY_ERROR_TEMP_3176")
Review Comment:
nit: sorry you need to define another error class; it's saying applyInPandasWithState. OK to add another temporary error class as we want to fix it anyway.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-49676][SS][PYTHON] Add Support for Chaining of Operators in transformWithStateInPandas API [spark]
Posted by "jingz-db (via GitHub)" <gi...@apache.org>.
jingz-db commented on code in PR #48124:
URL: https://github.com/apache/spark/pull/48124#discussion_r1837300644
##########
sql/core/src/main/scala/org/apache/spark/sql/RelationalGroupedDataset.scala:
##########
@@ -529,7 +530,31 @@ class RelationalGroupedDataset protected[sql](
initialStateSchema = initialState.df.schema
)
}
- Dataset.ofRows(df.sparkSession, plan)
+
+ if (eventTimeColumnName.isEmpty) {
+ Dataset.ofRows(df.sparkSession, plan)
+ } else {
+ updateEventTimeColumnAfterTransformWithState(plan, eventTimeColumnName)
+ }
+ }
+
+ /**
+ * Creates a new dataset with updated eventTimeColumn after the transformWithState
+ * logical node.
+ */
+ private def updateEventTimeColumnAfterTransformWithState(
+ transformWithStateInPandas: LogicalPlan,
+ eventTimeColumnName: String): DataFrame = {
+ val transformWithStateDataset = Dataset.ofRows(
+ df.sparkSession,
+ transformWithStateInPandas
+ )
+
+ Dataset.ofRows(df.sparkSession, EliminateEventTimeWatermark(
Review Comment:
> EliminateEventTimeWatermark is no longer needed as we now have this in Analyzer rule.
Is this already done? I saw we still keep the TODO here: https://github.com/apache/spark/blob/d6b733492d3e1643babbcd2b1860e35af7928906/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/Analyzer.scala#L3933
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-49676][SS][PYTHON] Add Support for Chaining of Operators in transformWithStateInPandas API [spark]
Posted by "jingz-db (via GitHub)" <gi...@apache.org>.
jingz-db commented on PR #48124:
URL: https://github.com/apache/spark/pull/48124#issuecomment-2480233401
Hey @HeartSaVioR, could you take another look? Thanks!
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-49676][SS][PYTHON] Add Support for Chaining of Operators in transformWithStateInPandas API [spark]
Posted by "HeartSaVioR (via GitHub)" <gi...@apache.org>.
HeartSaVioR commented on PR #48124:
URL: https://github.com/apache/spark/pull/48124#issuecomment-2503010989
Thanks! Merging to master.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-49676][SS][PYTHON] Add Support for Chaining of Operators in transformWithStateInPandas API [spark]
Posted by "jingz-db (via GitHub)" <gi...@apache.org>.
jingz-db commented on code in PR #48124:
URL: https://github.com/apache/spark/pull/48124#discussion_r1837303580
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/IncrementalExecution.scala:
##########
@@ -439,6 +439,25 @@ class IncrementalExecution(
eventTimeWatermarkForEviction = iwEviction)
))
+ // UpdateEventTimeColumnExec is used to tag the eventTime column, and validate
+ // emitted rows adhere to watermark in the output of transformWithStateInp.
+ // Hence, this node shares the same watermark value as TransformWithStateInPandasExec.
+ // This is the same as above in TransformWithStateExec.
+ // The only difference is TransformWithStateInPandasExec is analysed slightly different
+ // in the physical plan with `Project` and it is not `SerializeFromObjectExec`.
+ case UpdateEventTimeColumnExec(eventTime, delay, None,
+ ProjectExec(projectList, t: TransformWithStateInPandasExec))
Review Comment:
This is a minor difference on how the logical operator is constructed.
For Scala, the logicalplan node is defined and constructed here as `CatalystSerde.serialize[U](mapped)`: https://github.com/apache/spark/blob/5e49665ac39b49b875d6970f93df59aedd830fa5/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/plans/logical/object.scala#L632
However for Python, the logical plan node is constructed without this CatalystSerde because it was not originally typed when it got to the `RelatonalGroupedDataset`, we need to explicitly resolve the output type by projection to the user defined schema of output:
https://github.com/apache/spark/blob/d1544907a1cf94ebd6bd70e78def0efa4d6d549d/sql/core/src/main/scala/org/apache/spark/sql/RelationalGroupedDataset.scala#L499
The output we got from the projectList variable here is the output schema defined by the user in the `outputStructType`:
> projectList: List(outputTimestamp#15)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-49676][SS][PYTHON] Add Support for Chaining of Operators in transformWithStateInPandas API [spark]
Posted by "jingz-db (via GitHub)" <gi...@apache.org>.
jingz-db commented on code in PR #48124:
URL: https://github.com/apache/spark/pull/48124#discussion_r1819740895
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/IncrementalExecution.scala:
##########
@@ -425,6 +425,29 @@ class IncrementalExecution(
eventTimeWatermarkForEviction = iwEviction)
))
+ case UpdateEventTimeColumnExec(eventTime, delay, None,
Review Comment:
The main difference is where we wrap the logical node. For Scala, we wrap the event time column and output a dataset type [here](https://github.com/apache/spark/pull/45376/files#diff-3437bb4bcaf2e18c305978985e474daab11e397dc5f4666c13c8e11da0d7180bR706). For Python, we wrap the event time column and returns a Dataframe type [here](https://github.com/apache/spark/pull/48124/files#diff-32b043dfe6b906fb2b240e8557f98b03a648ba792ce58d11b631744b34bcea71R481). So that results in a slightly difference in the physical plan when the query enters the `IncrementalExecution` case match here. I'll update this in the comment as well.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-49676][SS][PYTHON] Add Support for Chaining of Operators in transformWithStateInPandas API [spark]
Posted by "jingz-db (via GitHub)" <gi...@apache.org>.
jingz-db commented on code in PR #48124:
URL: https://github.com/apache/spark/pull/48124#discussion_r1819740895
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/IncrementalExecution.scala:
##########
@@ -425,6 +425,29 @@ class IncrementalExecution(
eventTimeWatermarkForEviction = iwEviction)
))
+ case UpdateEventTimeColumnExec(eventTime, delay, None,
Review Comment:
The main difference is where we wrap the logical node. For Scala, we wrap the event time column and output a dataset type [here](https://github.com/apache/spark/pull/45376/files#diff-3437bb4bcaf2e18c305978985e474daab11e397dc5f4666c13c8e11da0d7180bR706). For Python, we wrap the event time column and returns a Dataframe type [here](https://github.com/apache/spark/pull/48124/files#diff-32b043dfe6b906fb2b240e8557f98b03a648ba792ce58d11b631744b34bcea71R481). So that results in a slightly difference in the physical plan when the query enters the `IncrementalExecution` case match here. I'll update this in the comment as well.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-49676][SS][PYTHON] Add Support for Chaining of Operators in transformWithStateInPandas API [spark]
Posted by "jingz-db (via GitHub)" <gi...@apache.org>.
jingz-db commented on code in PR #48124:
URL: https://github.com/apache/spark/pull/48124#discussion_r1837406346
##########
python/pyspark/sql/tests/pandas/test_pandas_transform_with_state.py:
##########
@@ -542,15 +554,85 @@ def check_results(batch_df, batch_id):
Row(id="a", timestamp="20"),
Row(id="a-expired", timestamp="0"),
}
- else:
+ elif batch_id == 2:
# watermark has not progressed, so timer registered in batch 1(watermark = 10)
# has not yet expired
assert set(batch_df.sort("id").collect()) == {Row(id="a", timestamp="15")}
+ else:
+ for q in self.spark.streams.active:
+ q.stop()
self._test_transform_with_state_in_pandas_event_time(
EventTimeStatefulProcessor(), check_results
)
+ def _test_transform_with_state_in_pandas_chaining_ops(
Review Comment:
Yeah... It got lost in a merge conflicts. I just added back.
##########
python/pyspark/sql/tests/pandas/test_pandas_transform_with_state.py:
##########
@@ -542,15 +554,85 @@ def check_results(batch_df, batch_id):
Row(id="a", timestamp="20"),
Row(id="a-expired", timestamp="0"),
}
- else:
+ elif batch_id == 2:
# watermark has not progressed, so timer registered in batch 1(watermark = 10)
# has not yet expired
assert set(batch_df.sort("id").collect()) == {Row(id="a", timestamp="15")}
+ else:
+ for q in self.spark.streams.active:
+ q.stop()
self._test_transform_with_state_in_pandas_event_time(
EventTimeStatefulProcessor(), check_results
)
+ def _test_transform_with_state_in_pandas_chaining_ops(
Review Comment:
Yeah... It got lost in a resolve of merge conflicts. I just added back.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-49676][SS][PYTHON] Add Support for Chaining of Operators in transformWithStateInPandas API [spark]
Posted by "HeartSaVioR (via GitHub)" <gi...@apache.org>.
HeartSaVioR commented on code in PR #48124:
URL: https://github.com/apache/spark/pull/48124#discussion_r1837306856
##########
sql/core/src/main/scala/org/apache/spark/sql/RelationalGroupedDataset.scala:
##########
@@ -529,7 +530,31 @@ class RelationalGroupedDataset protected[sql](
initialStateSchema = initialState.df.schema
)
}
- Dataset.ofRows(df.sparkSession, plan)
+
+ if (eventTimeColumnName.isEmpty) {
+ Dataset.ofRows(df.sparkSession, plan)
+ } else {
+ updateEventTimeColumnAfterTransformWithState(plan, eventTimeColumnName)
+ }
+ }
+
+ /**
+ * Creates a new dataset with updated eventTimeColumn after the transformWithState
+ * logical node.
+ */
+ private def updateEventTimeColumnAfterTransformWithState(
+ transformWithStateInPandas: LogicalPlan,
+ eventTimeColumnName: String): DataFrame = {
+ val transformWithStateDataset = Dataset.ofRows(
+ df.sparkSession,
+ transformWithStateInPandas
+ )
+
+ Dataset.ofRows(df.sparkSession, EliminateEventTimeWatermark(
Review Comment:
Yeah I forgot to remove the TODO. My bad. It's added.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-49676][SS][PYTHON] Add Support for Chaining of Operators in transformWithStateInPandas API [spark]
Posted by "jingz-db (via GitHub)" <gi...@apache.org>.
jingz-db commented on code in PR #48124:
URL: https://github.com/apache/spark/pull/48124#discussion_r1823419459
##########
python/pyspark/sql/tests/pandas/test_pandas_transform_with_state.py:
##########
@@ -304,6 +306,93 @@ def check_results(batch_df, batch_id):
finally:
input_dir.cleanup()
+ def _test_transform_with_state_in_pandas_chaining_ops(
+ self, stateful_processor, check_results, timeMode="None"
+ ):
+ import pyspark.sql.functions as f
+
+ input_path = tempfile.mkdtemp()
+ self._prepare_test_resource1(input_path)
+
+ def prepare_batch1(input_path):
+ with open(input_path + "/text-test3.txt", "w") as fw:
+ fw.write("a, 20\n")
+
+ def prepare_batch2(input_path):
+ with open(input_path + "/text-test1.txt", "w") as fw:
+ fw.write("a, 4\n")
+
+ def prepare_batch3(input_path):
+ with open(input_path + "/text-test2.txt", "w") as fw:
+ fw.write("a, 11\n")
+ fw.write("a, 13\n")
+ fw.write("a, 15\n")
+
+ prepare_batch1(input_path)
+ prepare_batch2(input_path)
+ prepare_batch3(input_path)
+
+ df = self._build_test_df(input_path)
+ df = df.select("id",
+ f.from_unixtime(f.col("temperature")).alias("eventTime").cast("timestamp")) \
+ .withWatermark("eventTime", "10 seconds")
+
+ for q in self.spark.streams.active:
+ q.stop()
+ self.assertTrue(df.isStreaming)
+
+ output_schema = StructType(
+ [
+ StructField("id", StringType(), True),
+ StructField("outputTimestamp", TimestampType(), True)
+ ]
+ )
+
+ q = (
+ df.groupBy("id")
+ .transformWithStateInPandas(
+ statefulProcessor=stateful_processor,
+ outputStructType=output_schema,
+ outputMode="Update",
+ timeMode=timeMode,
+ eventTimeColumnName="outputTimestamp"
+ )
+ .groupBy("outputTimestamp")
+ .count()
+ .writeStream.queryName("chaining_ops_query")
+ .foreachBatch(check_results)
+ .outputMode("update")
+ .start()
+ )
+
+ self.assertEqual(q.name, "chaining_ops_query")
+ self.assertTrue(q.isActive)
+ q.processAllAvailable()
+ q.awaitTermination(10)
Review Comment:
Thanks for leaving a comments on the state-v2 PR! Our test suite is not very large and 10 seconds should be quite sufficient to finish. I am also following the same set ups in the whole suite - you may find other test case is also using 10.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-49676][SS][PYTHON] Add Support for Chaining of Operators in transformWithStateInPandas API [spark]
Posted by "jingz-db (via GitHub)" <gi...@apache.org>.
jingz-db commented on code in PR #48124:
URL: https://github.com/apache/spark/pull/48124#discussion_r1837405539
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/IncrementalExecution.scala:
##########
@@ -439,6 +439,25 @@ class IncrementalExecution(
eventTimeWatermarkForEviction = iwEviction)
))
+ // UpdateEventTimeColumnExec is used to tag the eventTime column, and validate
+ // emitted rows adhere to watermark in the output of transformWithStateInp.
+ // Hence, this node shares the same watermark value as TransformWithStateInPandasExec.
+ // This is the same as above in TransformWithStateExec.
+ // The only difference is TransformWithStateInPandasExec is analysed slightly different
+ // in the physical plan with `Project` and it is not `SerializeFromObjectExec`.
+ case UpdateEventTimeColumnExec(eventTime, delay, None,
+ ProjectExec(projectList, t: TransformWithStateInPandasExec))
Review Comment:
The output we got from the projectList variable here is the output schema of the output event time column:
> projectList: List(outputTimestamp#15)
I really appreciate you bring the optimization thing up here. After reading your comments, I tried to figure out where this extra projection is coming from and it took me very long to figure out - we did not add this projection anywhere in any of the event time related change. So then I found this is coming from the pushed down projection from the chaining operator - I chained `groupBy`, but only groupBy on some of the output columns of the TWS output. I am not super familiar with optimization of analyzer, but I think eventually this projection is coming from the pushed down filter from the aggregator operator. So we will see the extra layer of Projection here. That being said, the match case class that contains the `Projection` is not sufficient.
I added another match case class above and I also added another test case to test on cases that matches the situation without the `Projection`.
I did a quick test on Scala side and it doesn't have this issue. It is because TWS node is wrapped by `SerializeFromObjectExec` and this probably does not match with some optimizer rule.
But I think the exact match on the `ProjectExec` is a bit hacky - there is probably more optimization rule I am not aware of. Do you think we should do a recursive match here in case TWS node is (recursively) wrapped by other operators besides `Project`?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-49676][SS][PYTHON] Add Support for Chaining of Operators in transformWithStateInPandas API [spark]
Posted by "HeartSaVioR (via GitHub)" <gi...@apache.org>.
HeartSaVioR commented on code in PR #48124:
URL: https://github.com/apache/spark/pull/48124#discussion_r1836131863
##########
python/pyspark/sql/tests/pandas/test_pandas_transform_with_state.py:
##########
@@ -542,15 +554,85 @@ def check_results(batch_df, batch_id):
Row(id="a", timestamp="20"),
Row(id="a-expired", timestamp="0"),
}
- else:
+ elif batch_id == 2:
# watermark has not progressed, so timer registered in batch 1(watermark = 10)
# has not yet expired
assert set(batch_df.sort("id").collect()) == {Row(id="a", timestamp="15")}
+ else:
+ for q in self.spark.streams.active:
+ q.stop()
self._test_transform_with_state_in_pandas_event_time(
EventTimeStatefulProcessor(), check_results
)
+ def _test_transform_with_state_in_pandas_chaining_ops(
Review Comment:
I don't see any caller for this. Have you missed to add caller?
##########
sql/core/src/main/scala/org/apache/spark/sql/RelationalGroupedDataset.scala:
##########
@@ -529,7 +530,31 @@ class RelationalGroupedDataset protected[sql](
initialStateSchema = initialState.df.schema
)
}
- Dataset.ofRows(df.sparkSession, plan)
+
+ if (eventTimeColumnName.isEmpty) {
+ Dataset.ofRows(df.sparkSession, plan)
+ } else {
+ updateEventTimeColumnAfterTransformWithState(plan, eventTimeColumnName)
+ }
+ }
+
+ /**
+ * Creates a new dataset with updated eventTimeColumn after the transformWithState
+ * logical node.
+ */
+ private def updateEventTimeColumnAfterTransformWithState(
+ transformWithStateInPandas: LogicalPlan,
+ eventTimeColumnName: String): DataFrame = {
+ val transformWithStateDataset = Dataset.ofRows(
+ df.sparkSession,
+ transformWithStateInPandas
+ )
+
+ Dataset.ofRows(df.sparkSession, EliminateEventTimeWatermark(
Review Comment:
EliminateEventTimeWatermark is no longer needed as we now have this in Analyzer rule.
But maybe we could deal with this in separate PR to address other cases as well.
##########
python/pyspark/sql/tests/pandas/test_pandas_transform_with_state.py:
##########
@@ -26,13 +26,7 @@
from pyspark import SparkConf
from pyspark.sql.functions import split
-from pyspark.sql.types import (
- StringType,
- StructType,
- StructField,
- Row,
- IntegerType,
-)
+from pyspark.sql.types import StringType, StructType, StructField, Row, IntegerType, TimestampType
Review Comment:
nit: probably linter may complain?
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/python/TransformWithStateInPandasExec.scala:
##########
@@ -191,8 +223,15 @@ case class TransformWithStateInPandasExec(
val updatesStartTimeNs = currentTimeNs
val (dedupAttributes, argOffsets) = resolveArgOffsets(child.output, groupingAttributes)
- val data =
- groupAndProject(dataIterator, groupingAttributes, child.output, dedupAttributes)
+ // If timeout is based on event time, then filter late data based on watermark
+ val filteredIter = watermarkPredicateForDataForLateEvents match {
Review Comment:
Wow, I wonder how we could miss this... Nice catch! I was about to ask for a new test covering this, but I'll revisit once you update the test as your new test seems to also contain this part.
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/IncrementalExecution.scala:
##########
@@ -439,6 +439,25 @@ class IncrementalExecution(
eventTimeWatermarkForEviction = iwEviction)
))
+ // UpdateEventTimeColumnExec is used to tag the eventTime column, and validate
+ // emitted rows adhere to watermark in the output of transformWithStateInp.
+ // Hence, this node shares the same watermark value as TransformWithStateInPandasExec.
+ // This is the same as above in TransformWithStateExec.
+ // The only difference is TransformWithStateInPandasExec is analysed slightly different
+ // in the physical plan with `Project` and it is not `SerializeFromObjectExec`.
+ case UpdateEventTimeColumnExec(eventTime, delay, None,
+ ProjectExec(projectList, t: TransformWithStateInPandasExec))
Review Comment:
I wonder why Project is necessary between `UpdateEventTimeColumn` and `TransformWithStateInPandas` - could you please check the value of projectList? If that value is the same with output of TransformWithStateInPandasExec, optimizer might be able to remove out Project (in future). If that's the case, we may want to address the pattern of not having a Project in between as well.
##########
python/pyspark/sql/tests/pandas/test_pandas_transform_with_state.py:
##########
@@ -241,11 +235,15 @@ def check_results(batch_df, _):
# test list state with ttl has the same behavior as list state when state doesn't expire.
def test_transform_with_state_in_pandas_list_state_large_ttl(self):
- def check_results(batch_df, _):
- assert set(batch_df.sort("id").collect()) == {
- Row(id="0", countAsString="2"),
- Row(id="1", countAsString="2"),
- }
+ def check_results(batch_df, batch_id):
+ if batch_id == 0:
+ assert set(batch_df.sort("id").collect()) == {
+ Row(id="0", countAsString="2"),
+ Row(id="1", countAsString="2"),
+ }
+ else:
Review Comment:
Instead of doing this in all places, shall we find the way for pytest to perform the same with before() / after()?
Also, since we are reading from files, why not just use Trigger.AvailableNow() which will terminate the query automatically instead of doing this? sleep(10) won't be necessary as well.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-49676][SS][PYTHON] Add Support for Chaining of Operators in transformWithStateInPandas API [spark]
Posted by "HeartSaVioR (via GitHub)" <gi...@apache.org>.
HeartSaVioR commented on code in PR #48124:
URL: https://github.com/apache/spark/pull/48124#discussion_r1857666519
##########
python/pyspark/sql/tests/pandas/test_pandas_transform_with_state.py:
##########
@@ -797,25 +777,22 @@ def check_results(batch_df, batch_id):
import datetime
if batch_id == 0:
- assert set(
- batch_df.sort("outputTimestamp").select("outputTimestamp", "count").collect()
- ) == {
- Row(outputTimestamp=datetime.datetime(1970, 1, 1, 0, 0, 20), count=1),
- }
+ assert batch_df.isEmpty()
elif batch_id == 1:
+ # watermark is 25 - 5 = 20, no more event for eventTime=10
Review Comment:
No, it's not.
* The watermark for "late events" in batch ID "N" is the watermark for "eviction" in batch "N - 1". (default: 0)
* The watermark for "eviction" in batch "N" is "calculated" based on the input data in batch "N - 1".
That said, the watermark for "eviction" in batch 1 is, 15 (max event time from batch 0) - 5 = 10. The watermark for "late events" in batch 0 is, 0.
With the same logic, the watermark for "eviction" in batch 2 is, 25 - 5 = 20. The watermark for "late events" in batch 2 is, the watermark for "eviction" in batch 1, hence 10.
Please update the code comment with the correction. I see both comments in batch 1 and 2 are incorrect.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-49676][SS][PYTHON] Add Support for Chaining of Operators in transformWithStateInPandas API [spark]
Posted by "jingz-db (via GitHub)" <gi...@apache.org>.
jingz-db commented on code in PR #48124:
URL: https://github.com/apache/spark/pull/48124#discussion_r1823449990
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/python/TransformWithStateInPandasExec.scala:
##########
@@ -106,6 +107,37 @@ case class TransformWithStateInPandasExec(
List.empty
}
+ override def shouldRunAnotherBatch(newInputWatermark: Long): Boolean = {
+ if (timeMode == ProcessingTime) {
+ // TODO: check if we can return true only if actual timers are registered, or there is
Review Comment:
> Since this PR is fairly light-weighted, should we include that check in the same PR?
Some additional testing on AvailableNow trigger mode may be needed together with fixing the issue and it exceeds the description of this PR. I'll leave it as it is today and let's keep track of the fix in the [SPARK-50180](https://issues.apache.org/jira/browse/SPARK-50180)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-49676][SS][PYTHON] Add Support for Chaining of Operators in transformWithStateInPandas API [spark]
Posted by "micheal-o (via GitHub)" <gi...@apache.org>.
micheal-o commented on code in PR #48124:
URL: https://github.com/apache/spark/pull/48124#discussion_r1762220932
##########
python/pyspark/sql/tests/pandas/test_pandas_transform_with_state.py:
##########
@@ -304,6 +306,93 @@ def check_results(batch_df, batch_id):
finally:
input_dir.cleanup()
+ def _test_transform_with_state_in_pandas_chaining_ops(
+ self, stateful_processor, check_results, timeMode="None"
+ ):
+ import pyspark.sql.functions as f
+
+ input_path = tempfile.mkdtemp()
+ self._prepare_test_resource1(input_path)
+
+ def prepare_batch1(input_path):
+ with open(input_path + "/text-test3.txt", "w") as fw:
+ fw.write("a, 20\n")
+
+ def prepare_batch2(input_path):
+ with open(input_path + "/text-test1.txt", "w") as fw:
+ fw.write("a, 4\n")
+
+ def prepare_batch3(input_path):
+ with open(input_path + "/text-test2.txt", "w") as fw:
+ fw.write("a, 11\n")
+ fw.write("a, 13\n")
+ fw.write("a, 15\n")
+
+ prepare_batch1(input_path)
+ prepare_batch2(input_path)
+ prepare_batch3(input_path)
+
+ df = self._build_test_df(input_path)
+ df = df.select("id",
+ f.from_unixtime(f.col("temperature")).alias("eventTime").cast("timestamp")) \
+ .withWatermark("eventTime", "10 seconds")
+
+ for q in self.spark.streams.active:
+ q.stop()
+ self.assertTrue(df.isStreaming)
+
+ output_schema = StructType(
+ [
+ StructField("id", StringType(), True),
+ StructField("outputTimestamp", TimestampType(), True)
+ ]
+ )
+
+ q = (
+ df.groupBy("id")
+ .transformWithStateInPandas(
+ statefulProcessor=stateful_processor,
+ outputStructType=output_schema,
+ outputMode="Update",
+ timeMode=timeMode,
+ eventTimeColumnName="outputTimestamp"
+ )
+ .groupBy("outputTimestamp")
+ .count()
+ .writeStream.queryName("chaining_ops_query")
+ .foreachBatch(check_results)
+ .outputMode("update")
+ .start()
+ )
+
+ self.assertEqual(q.name, "chaining_ops_query")
+ self.assertTrue(q.isActive)
+ q.processAllAvailable()
+ q.awaitTermination(10)
Review Comment:
nit: why wait for 10?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-49676][SS][PYTHON] Add Support for Chaining of Operators in transformWithStateInPandas API [spark]
Posted by "jingz-db (via GitHub)" <gi...@apache.org>.
jingz-db commented on code in PR #48124:
URL: https://github.com/apache/spark/pull/48124#discussion_r1837408393
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/python/TransformWithStateInPandasExec.scala:
##########
@@ -191,8 +223,15 @@ case class TransformWithStateInPandasExec(
val updatesStartTimeNs = currentTimeNs
val (dedupAttributes, argOffsets) = resolveArgOffsets(child.output, groupingAttributes)
- val data =
- groupAndProject(dataIterator, groupingAttributes, child.output, dedupAttributes)
+ // If timeout is based on event time, then filter late data based on watermark
+ val filteredIter = watermarkPredicateForDataForLateEvents match {
Review Comment:
Included a check on eviction of late events in the test case.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-49676][SS][PYTHON] Add Support for Chaining of Operators in transformWithStateInPandas API [spark]
Posted by "jingz-db (via GitHub)" <gi...@apache.org>.
jingz-db commented on code in PR #48124:
URL: https://github.com/apache/spark/pull/48124#discussion_r1842906748
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/IncrementalExecution.scala:
##########
@@ -439,6 +439,25 @@ class IncrementalExecution(
eventTimeWatermarkForEviction = iwEviction)
))
+ // UpdateEventTimeColumnExec is used to tag the eventTime column, and validate
+ // emitted rows adhere to watermark in the output of transformWithStateInp.
+ // Hence, this node shares the same watermark value as TransformWithStateInPandasExec.
+ // This is the same as above in TransformWithStateExec.
+ // The only difference is TransformWithStateInPandasExec is analysed slightly different
+ // in the physical plan with `Project` and it is not `SerializeFromObjectExec`.
+ case UpdateEventTimeColumnExec(eventTime, delay, None,
+ ProjectExec(projectList, t: TransformWithStateInPandasExec))
Review Comment:
Thanks for all the code pointers! I added a line in the `Optimizer` and the the column is now not pruned so UpdateEventTimeColumnExec and TransformWithStateInPandasExec are now placed together. I leave the rule inside Optimizer and `IncrementalExecution` will only match with only the case where UpdateEventTimeColumnExec and TransformWithStateInPandasExec are placed together with no extra layer in between.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-49676][SS][PYTHON] Add Support for Chaining of Operators in transformWithStateInPandas API [spark]
Posted by "HeartSaVioR (via GitHub)" <gi...@apache.org>.
HeartSaVioR commented on code in PR #48124:
URL: https://github.com/apache/spark/pull/48124#discussion_r1849453476
##########
python/pyspark/sql/tests/pandas/test_pandas_transform_with_state.py:
##########
@@ -701,6 +722,109 @@ def check_results(batch_df, batch_id):
SimpleStatefulProcessorWithInitialState(), check_results, initial_state
)
+ def _test_transform_with_state_in_pandas_chaining_ops(
+ self, stateful_processor, check_results, timeMode="None", grouping_cols=["outputTimestamp"]
+ ):
+ import pyspark.sql.functions as f
+
+ input_path = tempfile.mkdtemp()
+ self._prepare_test_resource1(input_path)
+
+ def prepare_batch1(input_path):
+ with open(input_path + "/text-test3.txt", "w") as fw:
+ fw.write("a, 20\n")
+
+ def prepare_batch2(input_path):
+ with open(input_path + "/text-test4.txt", "w") as fw:
+ fw.write("a, 3\n")
+
+ def prepare_batch3(input_path):
+ with open(input_path + "/text-test1.txt", "w") as fw:
+ fw.write("a, 4\n")
+
+ def prepare_batch4(input_path):
+ with open(input_path + "/text-test2.txt", "w") as fw:
+ fw.write("a, 20\n")
+
+ prepare_batch1(input_path)
+ time.sleep(2)
+ prepare_batch2(input_path)
+ time.sleep(2)
+ prepare_batch3(input_path)
+ time.sleep(2)
+ prepare_batch4(input_path)
+
+ df = self._build_test_df(input_path)
+ df = df.select(
+ "id", f.from_unixtime(f.col("temperature")).alias("eventTime").cast("timestamp")
+ ).withWatermark("eventTime", "10 seconds")
+
+ for q in self.spark.streams.active:
+ q.stop()
+ self.assertTrue(df.isStreaming)
+
+ output_schema = StructType(
+ [
+ StructField("id", StringType(), True),
+ StructField("outputTimestamp", TimestampType(), True),
+ ]
+ )
+
+ q = (
+ df.groupBy("id")
+ .transformWithStateInPandas(
+ statefulProcessor=stateful_processor,
+ outputStructType=output_schema,
+ outputMode="Update",
+ timeMode=timeMode,
+ eventTimeColumnName="outputTimestamp",
+ )
+ .groupBy(grouping_cols)
+ .count()
+ .writeStream.queryName("chaining_ops_query")
+ .foreachBatch(check_results)
+ .outputMode("update")
Review Comment:
update mode? We don't support multiple stateful operators with update mode. If the query does not break, we are missing whether the operator is stateful or not. We should check UnsupportedOperator.
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/Optimizer.scala:
##########
@@ -1031,6 +1031,9 @@ object ColumnPruning extends Rule[LogicalPlan] {
// Can't prune the columns on LeafNode
case p @ Project(_, _: LeafNode) => p
+ // Can't prune the columns on UpdateEventTimeWatermarkColumn
+ case p@Project(_, _: UpdateEventTimeWatermarkColumn) => p
Review Comment:
nit: space between p and `@`, and `P`
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-49676][SS][PYTHON] Add Support for Chaining of Operators in transformWithStateInPandas API [spark]
Posted by "jingz-db (via GitHub)" <gi...@apache.org>.
jingz-db commented on code in PR #48124:
URL: https://github.com/apache/spark/pull/48124#discussion_r1857529671
##########
python/pyspark/sql/tests/pandas/test_pandas_transform_with_state.py:
##########
@@ -701,6 +722,109 @@ def check_results(batch_df, batch_id):
SimpleStatefulProcessorWithInitialState(), check_results, initial_state
)
+ def _test_transform_with_state_in_pandas_chaining_ops(
+ self, stateful_processor, check_results, timeMode="None", grouping_cols=["outputTimestamp"]
+ ):
+ import pyspark.sql.functions as f
+
+ input_path = tempfile.mkdtemp()
+ self._prepare_test_resource1(input_path)
+
+ def prepare_batch1(input_path):
+ with open(input_path + "/text-test3.txt", "w") as fw:
+ fw.write("a, 20\n")
+
+ def prepare_batch2(input_path):
+ with open(input_path + "/text-test4.txt", "w") as fw:
+ fw.write("a, 3\n")
+
+ def prepare_batch3(input_path):
+ with open(input_path + "/text-test1.txt", "w") as fw:
+ fw.write("a, 4\n")
+
+ def prepare_batch4(input_path):
+ with open(input_path + "/text-test2.txt", "w") as fw:
+ fw.write("a, 20\n")
+
+ prepare_batch1(input_path)
+ time.sleep(2)
+ prepare_batch2(input_path)
+ time.sleep(2)
+ prepare_batch3(input_path)
+ time.sleep(2)
+ prepare_batch4(input_path)
+
+ df = self._build_test_df(input_path)
+ df = df.select(
+ "id", f.from_unixtime(f.col("temperature")).alias("eventTime").cast("timestamp")
+ ).withWatermark("eventTime", "10 seconds")
+
+ for q in self.spark.streams.active:
+ q.stop()
+ self.assertTrue(df.isStreaming)
+
+ output_schema = StructType(
+ [
+ StructField("id", StringType(), True),
+ StructField("outputTimestamp", TimestampType(), True),
+ ]
+ )
+
+ q = (
+ df.groupBy("id")
+ .transformWithStateInPandas(
+ statefulProcessor=stateful_processor,
+ outputStructType=output_schema,
+ outputMode="Append",
+ timeMode=timeMode,
+ eventTimeColumnName="outputTimestamp",
+ )
+ .groupBy(grouping_cols)
+ .count()
+ .writeStream.queryName("chaining_ops_query")
+ .foreachBatch(check_results)
+ .outputMode("append")
+ .start()
+ )
+
+ self.assertEqual(q.name, "chaining_ops_query")
+ self.assertTrue(q.isActive)
+ q.processAllAvailable()
+ q.awaitTermination(10)
+
+ def test_transform_with_state_in_pandas_chaining_ops(self):
+ def check_results(batch_df, batch_id):
+ import datetime
+
+ if batch_id == 0:
+ assert set(
+ batch_df.sort("outputTimestamp").select("outputTimestamp", "count").collect()
+ ) == {
+ Row(outputTimestamp=datetime.datetime(1970, 1, 1, 0, 0, 20), count=1),
+ }
+ elif batch_id == 1:
+ assert set(
+ batch_df.sort("outputTimestamp").select("outputTimestamp", "count").collect()
+ ) == {
+ Row(outputTimestamp=datetime.datetime(1970, 1, 1, 0, 0, 3), count=1),
+ }
+ elif batch_id == 2:
+ # as the late event watermark is 10, eventTime=3 is dropped
+ assert batch_df.isEmpty()
+ else:
+ assert set(
+ batch_df.sort("outputTimestamp").select("outputTimestamp", "count").collect()
+ ) == {
+ Row(outputTimestamp=datetime.datetime(1970, 1, 1, 0, 0, 20), count=2),
+ }
+
+ self._test_transform_with_state_in_pandas_chaining_ops(
+ StatefulProcessorChainingOps(), check_results, "eventTime"
+ )
+ self._test_transform_with_state_in_pandas_chaining_ops(
Review Comment:
The new test case now includes input of different ids. The purpose of this test is checking the column pruning is correctly set and the case class in `IncrementalExecution` matches with all logical plans.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-49676][SS][PYTHON] Add Support for Chaining of Operators in transformWithStateInPandas API [spark]
Posted by "HeartSaVioR (via GitHub)" <gi...@apache.org>.
HeartSaVioR commented on code in PR #48124:
URL: https://github.com/apache/spark/pull/48124#discussion_r1857666519
##########
python/pyspark/sql/tests/pandas/test_pandas_transform_with_state.py:
##########
@@ -797,25 +777,22 @@ def check_results(batch_df, batch_id):
import datetime
if batch_id == 0:
- assert set(
- batch_df.sort("outputTimestamp").select("outputTimestamp", "count").collect()
- ) == {
- Row(outputTimestamp=datetime.datetime(1970, 1, 1, 0, 0, 20), count=1),
- }
+ assert batch_df.isEmpty()
elif batch_id == 1:
+ # watermark is 25 - 5 = 20, no more event for eventTime=10
Review Comment:
No, it's not.
* The watermark for "late events" in batch ID "N" is the watermark for "eviction" in batch "N - 1". (default: 0)
* The watermark for "eviction" in batch "N" is "calculated" based on the input data in batch "N - 1".
That said, the watermark for "eviction" in batch 1 is, 15 (max event time from batch 0) - 5 = 10. The watermark for "late events" in batch 0 is, 0.
With the same logic, the watermark for "eviction" in batch 2 is, 25 - 5 = 20. The watermark for "late events" in batch 2 is, the watermark for "eviction" in batch 1, hence 10.
Please update the code comment with the correction. I see both comments in batch 1 and 2 are incorrect.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-49676][SS][PYTHON] Add Support for Chaining of Operators in transformWithStateInPandas API [spark]
Posted by "HeartSaVioR (via GitHub)" <gi...@apache.org>.
HeartSaVioR closed pull request #48124: [SPARK-49676][SS][PYTHON] Add Support for Chaining of Operators in transformWithStateInPandas API
URL: https://github.com/apache/spark/pull/48124
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-49676][SS][PYTHON] Add Support for Chaining of Operators in transformWithStateInPandas API [spark]
Posted by "HeartSaVioR (via GitHub)" <gi...@apache.org>.
HeartSaVioR commented on code in PR #48124:
URL: https://github.com/apache/spark/pull/48124#discussion_r1857677243
##########
python/pyspark/sql/tests/pandas/test_pandas_transform_with_state.py:
##########
@@ -701,6 +722,109 @@ def check_results(batch_df, batch_id):
SimpleStatefulProcessorWithInitialState(), check_results, initial_state
)
+ def _test_transform_with_state_in_pandas_chaining_ops(
+ self, stateful_processor, check_results, timeMode="None", grouping_cols=["outputTimestamp"]
+ ):
+ import pyspark.sql.functions as f
+
+ input_path = tempfile.mkdtemp()
+ self._prepare_test_resource1(input_path)
+
+ def prepare_batch1(input_path):
+ with open(input_path + "/text-test3.txt", "w") as fw:
+ fw.write("a, 20\n")
+
+ def prepare_batch2(input_path):
+ with open(input_path + "/text-test4.txt", "w") as fw:
+ fw.write("a, 3\n")
+
+ def prepare_batch3(input_path):
+ with open(input_path + "/text-test1.txt", "w") as fw:
+ fw.write("a, 4\n")
+
+ def prepare_batch4(input_path):
+ with open(input_path + "/text-test2.txt", "w") as fw:
+ fw.write("a, 20\n")
+
+ prepare_batch1(input_path)
+ time.sleep(2)
+ prepare_batch2(input_path)
+ time.sleep(2)
+ prepare_batch3(input_path)
+ time.sleep(2)
+ prepare_batch4(input_path)
+
+ df = self._build_test_df(input_path)
+ df = df.select(
+ "id", f.from_unixtime(f.col("temperature")).alias("eventTime").cast("timestamp")
+ ).withWatermark("eventTime", "10 seconds")
+
+ for q in self.spark.streams.active:
+ q.stop()
+ self.assertTrue(df.isStreaming)
+
+ output_schema = StructType(
+ [
+ StructField("id", StringType(), True),
+ StructField("outputTimestamp", TimestampType(), True),
+ ]
+ )
+
+ q = (
+ df.groupBy("id")
+ .transformWithStateInPandas(
+ statefulProcessor=stateful_processor,
+ outputStructType=output_schema,
+ outputMode="Append",
+ timeMode=timeMode,
+ eventTimeColumnName="outputTimestamp",
+ )
+ .groupBy(grouping_cols)
+ .count()
+ .writeStream.queryName("chaining_ops_query")
+ .foreachBatch(check_results)
+ .outputMode("append")
+ .start()
+ )
+
+ self.assertEqual(q.name, "chaining_ops_query")
+ self.assertTrue(q.isActive)
+ q.processAllAvailable()
+ q.awaitTermination(10)
+
+ def test_transform_with_state_in_pandas_chaining_ops(self):
+ def check_results(batch_df, batch_id):
+ import datetime
+
+ if batch_id == 0:
+ assert set(
+ batch_df.sort("outputTimestamp").select("outputTimestamp", "count").collect()
+ ) == {
+ Row(outputTimestamp=datetime.datetime(1970, 1, 1, 0, 0, 20), count=1),
+ }
+ elif batch_id == 1:
+ assert set(
+ batch_df.sort("outputTimestamp").select("outputTimestamp", "count").collect()
+ ) == {
+ Row(outputTimestamp=datetime.datetime(1970, 1, 1, 0, 0, 3), count=1),
+ }
+ elif batch_id == 2:
+ # as the late event watermark is 10, eventTime=3 is dropped
+ assert batch_df.isEmpty()
+ else:
+ assert set(
+ batch_df.sort("outputTimestamp").select("outputTimestamp", "count").collect()
+ ) == {
+ Row(outputTimestamp=datetime.datetime(1970, 1, 1, 0, 0, 20), count=2),
+ }
+
+ self._test_transform_with_state_in_pandas_chaining_ops(
+ StatefulProcessorChainingOps(), check_results, "eventTime"
+ )
+ self._test_transform_with_state_in_pandas_chaining_ops(
Review Comment:
OK, so it's not our main interest to verify that the groupBy will produce the aggregated result per grouping key.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-49676][SS][PYTHON] Add Support for Chaining of Operators in transformWithStateInPandas API [spark]
Posted by "jingz-db (via GitHub)" <gi...@apache.org>.
jingz-db commented on code in PR #48124:
URL: https://github.com/apache/spark/pull/48124#discussion_r1857527362
##########
python/pyspark/sql/tests/pandas/test_pandas_transform_with_state.py:
##########
@@ -701,6 +722,109 @@ def check_results(batch_df, batch_id):
SimpleStatefulProcessorWithInitialState(), check_results, initial_state
)
+ def _test_transform_with_state_in_pandas_chaining_ops(
+ self, stateful_processor, check_results, timeMode="None", grouping_cols=["outputTimestamp"]
+ ):
+ import pyspark.sql.functions as f
+
+ input_path = tempfile.mkdtemp()
+ self._prepare_test_resource1(input_path)
Review Comment:
Fixed in my new commits! The test suite was still under progress when you made the comments. Feel free to skip the older commits and review the newest change.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-49676][SS][PYTHON] Add Support for Chaining of Operators in transformWithStateInPandas API [spark]
Posted by "jingz-db (via GitHub)" <gi...@apache.org>.
jingz-db commented on code in PR #48124:
URL: https://github.com/apache/spark/pull/48124#discussion_r1823420567
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/IncrementalExecution.scala:
##########
@@ -425,6 +425,29 @@ class IncrementalExecution(
eventTimeWatermarkForEviction = iwEviction)
))
+ case UpdateEventTimeColumnExec(eventTime, delay, None,
Review Comment:
We don't. Deleted a case class and will add more comments.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-49676][SS][PYTHON] Add Support for Chaining of Operators in transformWithStateInPandas API [spark]
Posted by "HeartSaVioR (via GitHub)" <gi...@apache.org>.
HeartSaVioR commented on code in PR #48124:
URL: https://github.com/apache/spark/pull/48124#discussion_r1853244570
##########
python/pyspark/sql/tests/pandas/test_pandas_transform_with_state.py:
##########
@@ -701,6 +722,109 @@ def check_results(batch_df, batch_id):
SimpleStatefulProcessorWithInitialState(), check_results, initial_state
)
+ def _test_transform_with_state_in_pandas_chaining_ops(
+ self, stateful_processor, check_results, timeMode="None", grouping_cols=["outputTimestamp"]
+ ):
+ import pyspark.sql.functions as f
+
+ input_path = tempfile.mkdtemp()
+ self._prepare_test_resource1(input_path)
Review Comment:
nit: We should remove this. This is no-op because the file is overwritten later, but it is definitely confusing.
##########
python/pyspark/sql/tests/pandas/test_pandas_transform_with_state.py:
##########
@@ -701,6 +722,109 @@ def check_results(batch_df, batch_id):
SimpleStatefulProcessorWithInitialState(), check_results, initial_state
)
+ def _test_transform_with_state_in_pandas_chaining_ops(
+ self, stateful_processor, check_results, timeMode="None", grouping_cols=["outputTimestamp"]
+ ):
+ import pyspark.sql.functions as f
+
+ input_path = tempfile.mkdtemp()
+ self._prepare_test_resource1(input_path)
Review Comment:
```suggestion
```
##########
python/pyspark/sql/tests/pandas/test_pandas_transform_with_state.py:
##########
@@ -701,6 +722,109 @@ def check_results(batch_df, batch_id):
SimpleStatefulProcessorWithInitialState(), check_results, initial_state
)
+ def _test_transform_with_state_in_pandas_chaining_ops(
+ self, stateful_processor, check_results, timeMode="None", grouping_cols=["outputTimestamp"]
+ ):
+ import pyspark.sql.functions as f
+
+ input_path = tempfile.mkdtemp()
+ self._prepare_test_resource1(input_path)
+
+ def prepare_batch1(input_path):
+ with open(input_path + "/text-test3.txt", "w") as fw:
+ fw.write("a, 20\n")
+
+ def prepare_batch2(input_path):
+ with open(input_path + "/text-test4.txt", "w") as fw:
+ fw.write("a, 3\n")
+
+ def prepare_batch3(input_path):
+ with open(input_path + "/text-test1.txt", "w") as fw:
+ fw.write("a, 4\n")
+
+ def prepare_batch4(input_path):
+ with open(input_path + "/text-test2.txt", "w") as fw:
+ fw.write("a, 20\n")
+
+ prepare_batch1(input_path)
+ time.sleep(2)
+ prepare_batch2(input_path)
+ time.sleep(2)
+ prepare_batch3(input_path)
+ time.sleep(2)
+ prepare_batch4(input_path)
+
+ df = self._build_test_df(input_path)
+ df = df.select(
+ "id", f.from_unixtime(f.col("temperature")).alias("eventTime").cast("timestamp")
+ ).withWatermark("eventTime", "10 seconds")
+
+ for q in self.spark.streams.active:
+ q.stop()
+ self.assertTrue(df.isStreaming)
+
+ output_schema = StructType(
+ [
+ StructField("id", StringType(), True),
+ StructField("outputTimestamp", TimestampType(), True),
+ ]
+ )
+
+ q = (
+ df.groupBy("id")
+ .transformWithStateInPandas(
+ statefulProcessor=stateful_processor,
+ outputStructType=output_schema,
+ outputMode="Append",
+ timeMode=timeMode,
+ eventTimeColumnName="outputTimestamp",
+ )
+ .groupBy(grouping_cols)
+ .count()
+ .writeStream.queryName("chaining_ops_query")
+ .foreachBatch(check_results)
+ .outputMode("append")
+ .start()
+ )
+
+ self.assertEqual(q.name, "chaining_ops_query")
+ self.assertTrue(q.isActive)
+ q.processAllAvailable()
+ q.awaitTermination(10)
+
+ def test_transform_with_state_in_pandas_chaining_ops(self):
+ def check_results(batch_df, batch_id):
Review Comment:
The test result will be changed by the output mode. It's not the same with update mode. I expect the test to fail - you can't get any output because watermark does not pass through any of the timestamp.
It's not just changing the output mode - you need to revisit the test and reason about how it will work with append mode, and modify if necessary.
##########
python/pyspark/sql/tests/pandas/test_pandas_transform_with_state.py:
##########
@@ -701,6 +722,109 @@ def check_results(batch_df, batch_id):
SimpleStatefulProcessorWithInitialState(), check_results, initial_state
)
+ def _test_transform_with_state_in_pandas_chaining_ops(
+ self, stateful_processor, check_results, timeMode="None", grouping_cols=["outputTimestamp"]
+ ):
+ import pyspark.sql.functions as f
+
+ input_path = tempfile.mkdtemp()
+ self._prepare_test_resource1(input_path)
+
+ def prepare_batch1(input_path):
+ with open(input_path + "/text-test3.txt", "w") as fw:
+ fw.write("a, 20\n")
+
+ def prepare_batch2(input_path):
+ with open(input_path + "/text-test4.txt", "w") as fw:
+ fw.write("a, 3\n")
+
+ def prepare_batch3(input_path):
+ with open(input_path + "/text-test1.txt", "w") as fw:
+ fw.write("a, 4\n")
+
+ def prepare_batch4(input_path):
+ with open(input_path + "/text-test2.txt", "w") as fw:
+ fw.write("a, 20\n")
+
+ prepare_batch1(input_path)
+ time.sleep(2)
+ prepare_batch2(input_path)
+ time.sleep(2)
+ prepare_batch3(input_path)
+ time.sleep(2)
+ prepare_batch4(input_path)
+
+ df = self._build_test_df(input_path)
+ df = df.select(
+ "id", f.from_unixtime(f.col("temperature")).alias("eventTime").cast("timestamp")
+ ).withWatermark("eventTime", "10 seconds")
+
+ for q in self.spark.streams.active:
+ q.stop()
+ self.assertTrue(df.isStreaming)
+
+ output_schema = StructType(
+ [
+ StructField("id", StringType(), True),
+ StructField("outputTimestamp", TimestampType(), True),
+ ]
+ )
+
+ q = (
+ df.groupBy("id")
+ .transformWithStateInPandas(
+ statefulProcessor=stateful_processor,
+ outputStructType=output_schema,
+ outputMode="Append",
+ timeMode=timeMode,
+ eventTimeColumnName="outputTimestamp",
+ )
+ .groupBy(grouping_cols)
+ .count()
+ .writeStream.queryName("chaining_ops_query")
+ .foreachBatch(check_results)
+ .outputMode("append")
+ .start()
+ )
+
+ self.assertEqual(q.name, "chaining_ops_query")
+ self.assertTrue(q.isActive)
+ q.processAllAvailable()
+ q.awaitTermination(10)
+
+ def test_transform_with_state_in_pandas_chaining_ops(self):
+ def check_results(batch_df, batch_id):
+ import datetime
+
+ if batch_id == 0:
+ assert set(
+ batch_df.sort("outputTimestamp").select("outputTimestamp", "count").collect()
+ ) == {
+ Row(outputTimestamp=datetime.datetime(1970, 1, 1, 0, 0, 20), count=1),
+ }
+ elif batch_id == 1:
+ assert set(
+ batch_df.sort("outputTimestamp").select("outputTimestamp", "count").collect()
+ ) == {
+ Row(outputTimestamp=datetime.datetime(1970, 1, 1, 0, 0, 3), count=1),
+ }
+ elif batch_id == 2:
+ # as the late event watermark is 10, eventTime=3 is dropped
+ assert batch_df.isEmpty()
+ else:
+ assert set(
+ batch_df.sort("outputTimestamp").select("outputTimestamp", "count").collect()
+ ) == {
+ Row(outputTimestamp=datetime.datetime(1970, 1, 1, 0, 0, 20), count=2),
+ }
+
+ self._test_transform_with_state_in_pandas_chaining_ops(
+ StatefulProcessorChainingOps(), check_results, "eventTime"
+ )
+ self._test_transform_with_state_in_pandas_chaining_ops(
Review Comment:
nit: either add some data for another id with having another check_results, or remove this. This is not an exhaustive test anyway as it doesn't really test we group by id as well.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-49676][SS][PYTHON] Add Support for Chaining of Operators in transformWithStateInPandas API [spark]
Posted by "jingz-db (via GitHub)" <gi...@apache.org>.
jingz-db commented on code in PR #48124:
URL: https://github.com/apache/spark/pull/48124#discussion_r1837410252
##########
python/pyspark/sql/tests/pandas/test_pandas_transform_with_state.py:
##########
@@ -241,11 +235,15 @@ def check_results(batch_df, _):
# test list state with ttl has the same behavior as list state when state doesn't expire.
def test_transform_with_state_in_pandas_list_state_large_ttl(self):
- def check_results(batch_df, _):
- assert set(batch_df.sort("id").collect()) == {
- Row(id="0", countAsString="2"),
- Row(id="1", countAsString="2"),
- }
+ def check_results(batch_df, batch_id):
+ if batch_id == 0:
+ assert set(batch_df.sort("id").collect()) == {
+ Row(id="0", countAsString="2"),
+ Row(id="1", countAsString="2"),
+ }
+ else:
Review Comment:
Trigger.AvailableNow() will also keep making new batches... I had a discussion with Bo previously here: https://github.com/apache/spark/pull/48124#discussion_r1790883181
Not sure if we can do this in the in the `after()` as the query will keep making new batches because `shouldRunAnotherBatch` will always return True for processingTime mode. Scala side doesn't have the issue because we are using the `testStream`, `CheckAnswer` framework and it will shut down the query explicitly.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-49676][SS][PYTHON] Add Support for Chaining of Operators in transformWithStateInPandas API [spark]
Posted by "jingz-db (via GitHub)" <gi...@apache.org>.
jingz-db commented on code in PR #48124:
URL: https://github.com/apache/spark/pull/48124#discussion_r1857529671
##########
python/pyspark/sql/tests/pandas/test_pandas_transform_with_state.py:
##########
@@ -701,6 +722,109 @@ def check_results(batch_df, batch_id):
SimpleStatefulProcessorWithInitialState(), check_results, initial_state
)
+ def _test_transform_with_state_in_pandas_chaining_ops(
+ self, stateful_processor, check_results, timeMode="None", grouping_cols=["outputTimestamp"]
+ ):
+ import pyspark.sql.functions as f
+
+ input_path = tempfile.mkdtemp()
+ self._prepare_test_resource1(input_path)
+
+ def prepare_batch1(input_path):
+ with open(input_path + "/text-test3.txt", "w") as fw:
+ fw.write("a, 20\n")
+
+ def prepare_batch2(input_path):
+ with open(input_path + "/text-test4.txt", "w") as fw:
+ fw.write("a, 3\n")
+
+ def prepare_batch3(input_path):
+ with open(input_path + "/text-test1.txt", "w") as fw:
+ fw.write("a, 4\n")
+
+ def prepare_batch4(input_path):
+ with open(input_path + "/text-test2.txt", "w") as fw:
+ fw.write("a, 20\n")
+
+ prepare_batch1(input_path)
+ time.sleep(2)
+ prepare_batch2(input_path)
+ time.sleep(2)
+ prepare_batch3(input_path)
+ time.sleep(2)
+ prepare_batch4(input_path)
+
+ df = self._build_test_df(input_path)
+ df = df.select(
+ "id", f.from_unixtime(f.col("temperature")).alias("eventTime").cast("timestamp")
+ ).withWatermark("eventTime", "10 seconds")
+
+ for q in self.spark.streams.active:
+ q.stop()
+ self.assertTrue(df.isStreaming)
+
+ output_schema = StructType(
+ [
+ StructField("id", StringType(), True),
+ StructField("outputTimestamp", TimestampType(), True),
+ ]
+ )
+
+ q = (
+ df.groupBy("id")
+ .transformWithStateInPandas(
+ statefulProcessor=stateful_processor,
+ outputStructType=output_schema,
+ outputMode="Append",
+ timeMode=timeMode,
+ eventTimeColumnName="outputTimestamp",
+ )
+ .groupBy(grouping_cols)
+ .count()
+ .writeStream.queryName("chaining_ops_query")
+ .foreachBatch(check_results)
+ .outputMode("append")
+ .start()
+ )
+
+ self.assertEqual(q.name, "chaining_ops_query")
+ self.assertTrue(q.isActive)
+ q.processAllAvailable()
+ q.awaitTermination(10)
+
+ def test_transform_with_state_in_pandas_chaining_ops(self):
+ def check_results(batch_df, batch_id):
+ import datetime
+
+ if batch_id == 0:
+ assert set(
+ batch_df.sort("outputTimestamp").select("outputTimestamp", "count").collect()
+ ) == {
+ Row(outputTimestamp=datetime.datetime(1970, 1, 1, 0, 0, 20), count=1),
+ }
+ elif batch_id == 1:
+ assert set(
+ batch_df.sort("outputTimestamp").select("outputTimestamp", "count").collect()
+ ) == {
+ Row(outputTimestamp=datetime.datetime(1970, 1, 1, 0, 0, 3), count=1),
+ }
+ elif batch_id == 2:
+ # as the late event watermark is 10, eventTime=3 is dropped
+ assert batch_df.isEmpty()
+ else:
+ assert set(
+ batch_df.sort("outputTimestamp").select("outputTimestamp", "count").collect()
+ ) == {
+ Row(outputTimestamp=datetime.datetime(1970, 1, 1, 0, 0, 20), count=2),
+ }
+
+ self._test_transform_with_state_in_pandas_chaining_ops(
+ StatefulProcessorChainingOps(), check_results, "eventTime"
+ )
+ self._test_transform_with_state_in_pandas_chaining_ops(
Review Comment:
The new test case now includes input of different ids. The purpose of this test was checking the column pruning is correctly set and the case class in `IncrementalExecution` matches with all logical plans.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-49676][SS][PYTHON] Add Support for Chaining of Operators in transformWithStateInPandas API [spark]
Posted by "jingz-db (via GitHub)" <gi...@apache.org>.
jingz-db commented on code in PR #48124:
URL: https://github.com/apache/spark/pull/48124#discussion_r1837416816
##########
python/pyspark/sql/tests/pandas/test_pandas_transform_with_state.py:
##########
@@ -26,13 +26,7 @@
from pyspark import SparkConf
from pyspark.sql.functions import split
-from pyspark.sql.types import (
- StringType,
- StructType,
- StructField,
- Row,
- IntegerType,
-)
+from pyspark.sql.types import StringType, StructType, StructField, Row, IntegerType, TimestampType
Review Comment:
Ran the linter check and seems not complaining... If it is OK with you, I'll leave it here and see if it passes on the CI.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-49676][SS][PYTHON] Add Support for Chaining of Operators in transformWithStateInPandas API [spark]
Posted by "jingz-db (via GitHub)" <gi...@apache.org>.
jingz-db commented on code in PR #48124:
URL: https://github.com/apache/spark/pull/48124#discussion_r1859475047
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/SparkStrategies.scala:
##########
@@ -966,6 +966,9 @@ abstract class SparkStrategies extends QueryPlanner[SparkPlan] {
case _: FlatMapGroupsInPandasWithState =>
// TODO(SPARK-40443): support applyInPandasWithState in batch query
throw new SparkUnsupportedOperationException("_LEGACY_ERROR_TEMP_3176")
+ case _: TransformWithStateInPandas =>
+ // TODO(SPARK-50428): support TransformWithStateInPandas in batch query
+ throw new SparkUnsupportedOperationException("_LEGACY_ERROR_TEMP_3176")
Review Comment:
Reuse this error class used for `TransformWithState` previously: [https://github.com/apache/spark/commit/289543eba5bd07395282c28cb08934ec625a4935#diff-7c879c08d2f379c139d5229a888[…]69bb48f0a138a3d64e1b2dde3502feR46](https://github.com/apache/spark/commit/289543eba5bd07395282c28cb08934ec625a4935#diff-7c879c08d2f379c139d5229a88857229ae69bb48f0a138a3d64e1b2dde3502feR46)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-49676][SS][PYTHON] Add Support for Chaining of Operators in transformWithStateInPandas API [spark]
Posted by "jingz-db (via GitHub)" <gi...@apache.org>.
jingz-db commented on code in PR #48124:
URL: https://github.com/apache/spark/pull/48124#discussion_r1823433041
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/python/TransformWithStateInPandasExec.scala:
##########
@@ -106,6 +107,37 @@ case class TransformWithStateInPandasExec(
List.empty
}
+ override def shouldRunAnotherBatch(newInputWatermark: Long): Boolean = {
+ if (timeMode == ProcessingTime) {
+ // TODO: check if we can return true only if actual timers are registered, or there is
Review Comment:
Confirmed with Anish - the way we have it today is that we'll keep constructing new batches because we may have future timer expiring. I sent out a SPARK JIRA to keep track of this issue: https://issues.apache.org/jira/browse/SPARK-50180 and will update the comments in both Scala and Python.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-49676][SS][PYTHON] Add Support for Chaining of Operators in transformWithStateInPandas API [spark]
Posted by "jingz-db (via GitHub)" <gi...@apache.org>.
jingz-db commented on code in PR #48124:
URL: https://github.com/apache/spark/pull/48124#discussion_r1837303580
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/IncrementalExecution.scala:
##########
@@ -439,6 +439,25 @@ class IncrementalExecution(
eventTimeWatermarkForEviction = iwEviction)
))
+ // UpdateEventTimeColumnExec is used to tag the eventTime column, and validate
+ // emitted rows adhere to watermark in the output of transformWithStateInp.
+ // Hence, this node shares the same watermark value as TransformWithStateInPandasExec.
+ // This is the same as above in TransformWithStateExec.
+ // The only difference is TransformWithStateInPandasExec is analysed slightly different
+ // in the physical plan with `Project` and it is not `SerializeFromObjectExec`.
+ case UpdateEventTimeColumnExec(eventTime, delay, None,
+ ProjectExec(projectList, t: TransformWithStateInPandasExec))
Review Comment:
This is a minor difference on how the logical operator is constructed.
For Scala, the logicalplan node is defined and constructed here as `CatalystSerde.serialize[U](mapped)`: https://github.com/apache/spark/blob/5e49665ac39b49b875d6970f93df59aedd830fa5/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/plans/logical/object.scala#L632
However for Python, the logical plan node is constructed without this CatalystSerde because it was not originally typed when it got to the `RelatonalGroupedDataset`, we need to explicitly resolve the output type by projection to the user defined schema of output:
https://github.com/apache/spark/blob/d1544907a1cf94ebd6bd70e78def0efa4d6d549d/sql/core/src/main/scala/org/apache/spark/sql/RelationalGroupedDataset.scala#L499
The output we got from the projectList variable here is the output schema defined by the user in the `outputStructType`:
> projectList: List(outputTimestamp#15)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-49676][SS][PYTHON] Add Support for Chaining of Operators in transformWithStateInPandas API [spark]
Posted by "HeartSaVioR (via GitHub)" <gi...@apache.org>.
HeartSaVioR commented on code in PR #48124:
URL: https://github.com/apache/spark/pull/48124#discussion_r1837434759
##########
python/pyspark/sql/tests/pandas/test_pandas_transform_with_state.py:
##########
@@ -241,11 +235,15 @@ def check_results(batch_df, _):
# test list state with ttl has the same behavior as list state when state doesn't expire.
def test_transform_with_state_in_pandas_list_state_large_ttl(self):
- def check_results(batch_df, _):
- assert set(batch_df.sort("id").collect()) == {
- Row(id="0", countAsString="2"),
- Row(id="1", countAsString="2"),
- }
+ def check_results(batch_df, batch_id):
+ if batch_id == 0:
+ assert set(batch_df.sort("id").collect()) == {
+ Row(id="0", countAsString="2"),
+ Row(id="1", countAsString="2"),
+ }
+ else:
Review Comment:
It is only happening with processing time timer mode and I don't expect a lot of tests would use that.
But I'm fine either way. I'm OK to use the same approach for all timer modes.
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/IncrementalExecution.scala:
##########
@@ -439,6 +439,25 @@ class IncrementalExecution(
eventTimeWatermarkForEviction = iwEviction)
))
+ // UpdateEventTimeColumnExec is used to tag the eventTime column, and validate
+ // emitted rows adhere to watermark in the output of transformWithStateInp.
+ // Hence, this node shares the same watermark value as TransformWithStateInPandasExec.
+ // This is the same as above in TransformWithStateExec.
+ // The only difference is TransformWithStateInPandasExec is analysed slightly different
+ // in the physical plan with `Project` and it is not `SerializeFromObjectExec`.
+ case UpdateEventTimeColumnExec(eventTime, delay, None,
+ ProjectExec(projectList, t: TransformWithStateInPandasExec))
Review Comment:
It's probably easier to not allow pushdown between UpdateEventTimeColumnExec and TransformWithStateExec.
I see filter is not allowed to be pushed down below UpdateEventTimeColumnExec, so it is OK. This might be the case of column pruning -
https://github.com/apache/spark/blob/master/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/Optimizer.scala#L1036-L1045
This could insert the project down to children nodes; we can disallow this for UpdateEventTimeColumnExec, like following:
https://github.com/apache/spark/blob/master/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/Optimizer.scala#L1031-L1032
Could you give a try with this and see you don't see Projection? If that works, we can just consider UpdateEventTimeColumnExec and TransformWithStateExec to be always close to each other.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-49676][SS][PYTHON] Add Support for Chaining of Operators in transformWithStateInPandas API [spark]
Posted by "jingz-db (via GitHub)" <gi...@apache.org>.
jingz-db commented on code in PR #48124:
URL: https://github.com/apache/spark/pull/48124#discussion_r1819737427
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/python/TransformWithStateInPandasExec.scala:
##########
@@ -106,6 +107,37 @@ case class TransformWithStateInPandasExec(
List.empty
}
+ override def shouldRunAnotherBatch(newInputWatermark: Long): Boolean = {
+ if (timeMode == ProcessingTime) {
+ // TODO: check if we can return true only if actual timers are registered, or there is
Review Comment:
This is also the way it is on Scala side [here](https://github.com/apache/spark/blob/738dfa346bde59d10fa18bc662b858b0c3354254/sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/TransformWithStateExec.scala#L88). Let me check if there is any specific reason for Bhuwan to leave this as a TODO here.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-49676][SS][PYTHON] Add Support for Chaining of Operators in transformWithStateInPandas API [spark]
Posted by "HeartSaVioR (via GitHub)" <gi...@apache.org>.
HeartSaVioR commented on code in PR #48124:
URL: https://github.com/apache/spark/pull/48124#discussion_r1857611909
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/UnsupportedOperationChecker.scala:
##########
@@ -44,6 +44,9 @@ object UnsupportedOperationChecker extends Logging {
throwError("dropDuplicatesWithinWatermark is not supported with batch " +
"DataFrames/DataSets")(d)
+ case t: TransformWithStateInPandas =>
Review Comment:
Let's file a JIRA ticket with TODO like this.
```
// TODO(SPARK-40443): support applyInPandasWithState in batch query
throw new SparkUnsupportedOperationException("_LEGACY_ERROR_TEMP_3176")
```
I see my inconsistency here - the above is in BasicOperators in SparkStrategies.
https://github.com/apache/spark/blob/f7122137006e941393c8be619fb51b3b713a24cb/sql/core/src/main/scala/org/apache/spark/sql/execution/SparkStrategies.scala#L966-L968
Shall we move this to BasicOperators with adding TODO comment like the above? Let's leave this place to be something there is a reason not to support and won't support in future.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-49676][SS][PYTHON] Add Support for Chaining of Operators in transformWithStateInPandas API [spark]
Posted by "jingz-db (via GitHub)" <gi...@apache.org>.
jingz-db commented on PR #48124:
URL: https://github.com/apache/spark/pull/48124#issuecomment-2502432382
> +1 pending CI. Thanks for the patience.
>
> I know it's a tech debt to leave this to temp error class, so let's file a JIRA ticket to change the error class to the non-temp one if we can't find the time to address the support of batch query.
Created here: https://issues.apache.org/jira/browse/SPARK-50429
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-49676][SS][PYTHON] Add Support for Chaining of Operators in transformWithStateInPandas API [spark]
Posted by "HeartSaVioR (via GitHub)" <gi...@apache.org>.
HeartSaVioR commented on code in PR #48124:
URL: https://github.com/apache/spark/pull/48124#discussion_r1853231861
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/UnsupportedOperationChecker.scala:
##########
@@ -103,6 +103,7 @@ object UnsupportedOperationChecker extends Logging {
case d: Deduplicate if d.isStreaming && d.keys.exists(hasEventTimeCol) => true
case d: DeduplicateWithinWatermark if d.isStreaming => true
case t: TransformWithState if t.isStreaming => true
+ case t: TransformWithStateInPandas if t.isStreaming => true
Review Comment:
Thanks for adding this. It is lucky we figured out before releasing the feature. :)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-49676][SS][PYTHON] Add Support for Chaining of Operators in transformWithStateInPandas API [spark]
Posted by "jingz-db (via GitHub)" <gi...@apache.org>.
jingz-db commented on code in PR #48124:
URL: https://github.com/apache/spark/pull/48124#discussion_r1859475047
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/SparkStrategies.scala:
##########
@@ -966,6 +966,9 @@ abstract class SparkStrategies extends QueryPlanner[SparkPlan] {
case _: FlatMapGroupsInPandasWithState =>
// TODO(SPARK-40443): support applyInPandasWithState in batch query
throw new SparkUnsupportedOperationException("_LEGACY_ERROR_TEMP_3176")
+ case _: TransformWithStateInPandas =>
+ // TODO(SPARK-50428): support TransformWithStateInPandas in batch query
+ throw new SparkUnsupportedOperationException("_LEGACY_ERROR_TEMP_3176")
Review Comment:
Modified to reuse this error class used for `TransformWithState` previously: [https://github.com/apache/spark/commit/289543eba5bd07395282c28cb08934ec625a4935#diff-7c879c08d2f379c139d5229a888[…]69bb48f0a138a3d64e1b2dde3502feR46](https://github.com/apache/spark/commit/289543eba5bd07395282c28cb08934ec625a4935#diff-7c879c08d2f379c139d5229a88857229ae69bb48f0a138a3d64e1b2dde3502feR46)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-49676][SS][PYTHON] Add Support for Chaining of Operators in transformWithStateInPandas API [spark]
Posted by "jingz-db (via GitHub)" <gi...@apache.org>.
jingz-db commented on code in PR #48124:
URL: https://github.com/apache/spark/pull/48124#discussion_r1859419007
##########
python/pyspark/sql/tests/pandas/test_pandas_transform_with_state.py:
##########
@@ -701,6 +722,109 @@ def check_results(batch_df, batch_id):
SimpleStatefulProcessorWithInitialState(), check_results, initial_state
)
+ def _test_transform_with_state_in_pandas_chaining_ops(
+ self, stateful_processor, check_results, timeMode="None", grouping_cols=["outputTimestamp"]
+ ):
+ import pyspark.sql.functions as f
+
+ input_path = tempfile.mkdtemp()
+ self._prepare_test_resource1(input_path)
+
+ def prepare_batch1(input_path):
+ with open(input_path + "/text-test3.txt", "w") as fw:
+ fw.write("a, 20\n")
+
+ def prepare_batch2(input_path):
+ with open(input_path + "/text-test4.txt", "w") as fw:
+ fw.write("a, 3\n")
+
+ def prepare_batch3(input_path):
+ with open(input_path + "/text-test1.txt", "w") as fw:
+ fw.write("a, 4\n")
+
+ def prepare_batch4(input_path):
+ with open(input_path + "/text-test2.txt", "w") as fw:
+ fw.write("a, 20\n")
+
+ prepare_batch1(input_path)
+ time.sleep(2)
+ prepare_batch2(input_path)
+ time.sleep(2)
+ prepare_batch3(input_path)
+ time.sleep(2)
+ prepare_batch4(input_path)
+
+ df = self._build_test_df(input_path)
+ df = df.select(
+ "id", f.from_unixtime(f.col("temperature")).alias("eventTime").cast("timestamp")
+ ).withWatermark("eventTime", "10 seconds")
+
+ for q in self.spark.streams.active:
+ q.stop()
+ self.assertTrue(df.isStreaming)
+
+ output_schema = StructType(
+ [
+ StructField("id", StringType(), True),
+ StructField("outputTimestamp", TimestampType(), True),
+ ]
+ )
+
+ q = (
+ df.groupBy("id")
+ .transformWithStateInPandas(
+ statefulProcessor=stateful_processor,
+ outputStructType=output_schema,
+ outputMode="Append",
+ timeMode=timeMode,
+ eventTimeColumnName="outputTimestamp",
+ )
+ .groupBy(grouping_cols)
+ .count()
+ .writeStream.queryName("chaining_ops_query")
+ .foreachBatch(check_results)
+ .outputMode("append")
+ .start()
+ )
+
+ self.assertEqual(q.name, "chaining_ops_query")
+ self.assertTrue(q.isActive)
+ q.processAllAvailable()
+ q.awaitTermination(10)
+
+ def test_transform_with_state_in_pandas_chaining_ops(self):
+ def check_results(batch_df, batch_id):
+ import datetime
+
+ if batch_id == 0:
+ assert set(
+ batch_df.sort("outputTimestamp").select("outputTimestamp", "count").collect()
+ ) == {
+ Row(outputTimestamp=datetime.datetime(1970, 1, 1, 0, 0, 20), count=1),
+ }
+ elif batch_id == 1:
+ assert set(
+ batch_df.sort("outputTimestamp").select("outputTimestamp", "count").collect()
+ ) == {
+ Row(outputTimestamp=datetime.datetime(1970, 1, 1, 0, 0, 3), count=1),
+ }
+ elif batch_id == 2:
+ # as the late event watermark is 10, eventTime=3 is dropped
+ assert batch_df.isEmpty()
+ else:
+ assert set(
+ batch_df.sort("outputTimestamp").select("outputTimestamp", "count").collect()
+ ) == {
+ Row(outputTimestamp=datetime.datetime(1970, 1, 1, 0, 0, 20), count=2),
+ }
+
+ self._test_transform_with_state_in_pandas_chaining_ops(
+ StatefulProcessorChainingOps(), check_results, "eventTime"
+ )
+ self._test_transform_with_state_in_pandas_chaining_ops(
Review Comment:
Yeah it is not our main interest as we are testing for event time watermark is propagated properly for TWS here. I believe we should have abundant tests for aggregation already. I am keeping this for safety because previously without the column pruning change, `groupby` on single column and multiple columns will match different case class in `IncrementalExecution`.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-49676][SS][PYTHON] Add Support for Chaining of Operators in transformWithStateInPandas API [spark]
Posted by "jingz-db (via GitHub)" <gi...@apache.org>.
jingz-db commented on code in PR #48124:
URL: https://github.com/apache/spark/pull/48124#discussion_r1857487373
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/UnsupportedOperationChecker.scala:
##########
@@ -103,6 +103,7 @@ object UnsupportedOperationChecker extends Logging {
case d: Deduplicate if d.isStreaming && d.keys.exists(hasEventTimeCol) => true
case d: DeduplicateWithinWatermark if d.isStreaming => true
case t: TransformWithState if t.isStreaming => true
+ case t: TransformWithStateInPandas if t.isStreaming => true
Review Comment:
Did a comparison with TWS Scala base PR and added a blocker for batch query. Thanks for the reminding!
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-49676][SS][PYTHON] Add Support for Chaining of Operators in transformWithStateInPandas API [spark]
Posted by "HeartSaVioR (via GitHub)" <gi...@apache.org>.
HeartSaVioR commented on code in PR #48124:
URL: https://github.com/apache/spark/pull/48124#discussion_r1857666519
##########
python/pyspark/sql/tests/pandas/test_pandas_transform_with_state.py:
##########
@@ -797,25 +777,22 @@ def check_results(batch_df, batch_id):
import datetime
if batch_id == 0:
- assert set(
- batch_df.sort("outputTimestamp").select("outputTimestamp", "count").collect()
- ) == {
- Row(outputTimestamp=datetime.datetime(1970, 1, 1, 0, 0, 20), count=1),
- }
+ assert batch_df.isEmpty()
elif batch_id == 1:
+ # watermark is 25 - 5 = 20, no more event for eventTime=10
Review Comment:
No, it's not.
* The watermark for "late events" in batch ID "N" is the watermark for "eviction" in batch "N - 1". (default: 0)
* The watermark for "eviction" in batch "N" is "calculated" based on the input data in batch "N - 1".
That said, the watermark for "eviction" in batch 1 is, 15 (max event time from batch 0) - 5 = 10. The watermark for "late events" in batch 0 is, 0.
With the same logic, the watermark for "eviction" in batch 2 is, 25 - 5 = 20. The watermark for "late events" in batch 2 is, the watermark for "eviction" in batch 1, hence 10.
Please update the code comment with the correction. I see both comments in batch 1 and 2 are incorrect, though somehow the result is the same.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-49676][SS][PYTHON] Add Support for Chaining of Operators in transformWithStateInPandas API [spark]
Posted by "jingz-db (via GitHub)" <gi...@apache.org>.
jingz-db commented on code in PR #48124:
URL: https://github.com/apache/spark/pull/48124#discussion_r1837405539
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/IncrementalExecution.scala:
##########
@@ -439,6 +439,25 @@ class IncrementalExecution(
eventTimeWatermarkForEviction = iwEviction)
))
+ // UpdateEventTimeColumnExec is used to tag the eventTime column, and validate
+ // emitted rows adhere to watermark in the output of transformWithStateInp.
+ // Hence, this node shares the same watermark value as TransformWithStateInPandasExec.
+ // This is the same as above in TransformWithStateExec.
+ // The only difference is TransformWithStateInPandasExec is analysed slightly different
+ // in the physical plan with `Project` and it is not `SerializeFromObjectExec`.
+ case UpdateEventTimeColumnExec(eventTime, delay, None,
+ ProjectExec(projectList, t: TransformWithStateInPandasExec))
Review Comment:
The output we got from the projectList variable here is the output schema of the output event time column:
> projectList: List(outputTimestamp#15)
I really appreciate you bring the optimization thing up here. After reading your comments, I tried to figure out where this extra projection is coming from and it took me very long to figure out - we did not add this projection anywhere in any of the event time related change. I found this is probably coming from the pushed down projection from the chaining operator. In the test case, `groupBy` is chained after TWS, and this groupBy only aggregate on some of the output columns of the TWS output. I am not super familiar with optimization of analyzer, but I think eventually this projection is coming from the pushed down filter from the aggregator operator. So we will see the extra layer of Projection here. That being said, the match case class that contains the `Projection` is not sufficient.
I added another match case class above and I also added another test case to test on cases that matches the situation without the `Projection`.
I did a quick test on Scala side and it doesn't have this issue. It is because TWS node is wrapped by `SerializeFromObjectExec` and this probably does not match with some optimizer rule.
But I think the exact match on the `ProjectExec` is a bit hacky - there is probably more optimization rule I am not aware of. Do you think we should do a recursive match here in case TWS node is (recursively) wrapped by other operators besides `Project`?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org