You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@spark.apache.org by ka...@apache.org on 2024/07/19 02:52:49 UTC
(spark) branch branch-3.4 updated: Revert "[SPARK-48934][SS] Python datetime types converted incorrectly for setting timeout in applyInPandasWithState"
This is an automated email from the ASF dual-hosted git repository.
kabhwan pushed a commit to branch branch-3.4
in repository https://gitbox.apache.org/repos/asf/spark.git
The following commit(s) were added to refs/heads/branch-3.4 by this push:
new f84f4766da03 Revert "[SPARK-48934][SS] Python datetime types converted incorrectly for setting timeout in applyInPandasWithState"
f84f4766da03 is described below
commit f84f4766da032d01ee0210a0ea7a9ee40bf559b9
Author: Jungtaek Lim <ka...@gmail.com>
AuthorDate: Fri Jul 19 11:52:34 2024 +0900
Revert "[SPARK-48934][SS] Python datetime types converted incorrectly for setting timeout in applyInPandasWithState"
This reverts commit 4239a58743edfc00fc0de8ad649a850e96c45a4a.
---
python/pyspark/sql/streaming/state.py | 24 +--
.../FlatMapGroupsInPandasWithStateSuite.scala | 166 ++++++++++-----------
2 files changed, 80 insertions(+), 110 deletions(-)
diff --git a/python/pyspark/sql/streaming/state.py b/python/pyspark/sql/streaming/state.py
index 95bfe830bd65..f0ac427cbea3 100644
--- a/python/pyspark/sql/streaming/state.py
+++ b/python/pyspark/sql/streaming/state.py
@@ -18,7 +18,7 @@ import datetime
import json
from typing import Tuple, Optional
-from pyspark.sql.types import Row, StructType, TimestampType
+from pyspark.sql.types import DateType, Row, StructType
from pyspark.sql.utils import has_numpy
__all__ = ["GroupState", "GroupStateTimeout"]
@@ -178,17 +178,7 @@ class GroupState:
)
if durationMs <= 0:
-<<<<<<< HEAD
raise ValueError("Timeout duration must be positive")
-=======
- raise PySparkValueError(
- error_class="VALUE_NOT_POSITIVE",
- message_parameters={
- "arg_name": "durationMs",
- "arg_value": type(durationMs).__name__,
- },
- )
->>>>>>> a54daa196c1 ([SPARK-48934][SS] Python datetime types converted incorrectly for setting timeout in applyInPandasWithState)
self._timeout_timestamp = durationMs + self._batch_processing_time_ms
# TODO(SPARK-40438): Implement additionalDuration parameter.
@@ -205,20 +195,10 @@ class GroupState:
)
if isinstance(timestampMs, datetime.datetime):
- timestampMs = TimestampType().toInternal(timestampMs) / 1000
+ timestampMs = DateType().toInternal(timestampMs)
if timestampMs <= 0:
-<<<<<<< HEAD
raise ValueError("Timeout timestamp must be positive")
-=======
- raise PySparkValueError(
- error_class="VALUE_NOT_POSITIVE",
- message_parameters={
- "arg_name": "timestampMs",
- "arg_value": type(timestampMs).__name__,
- },
- )
->>>>>>> a54daa196c1 ([SPARK-48934][SS] Python datetime types converted incorrectly for setting timeout in applyInPandasWithState)
if (
self._event_time_watermark_ms != GroupState.NO_TIMESTAMP
diff --git a/sql/core/src/test/scala/org/apache/spark/sql/streaming/FlatMapGroupsInPandasWithStateSuite.scala b/sql/core/src/test/scala/org/apache/spark/sql/streaming/FlatMapGroupsInPandasWithStateSuite.scala
index b3a4b8e3d3cb..20fb17fe6ec2 100644
--- a/sql/core/src/test/scala/org/apache/spark/sql/streaming/FlatMapGroupsInPandasWithStateSuite.scala
+++ b/sql/core/src/test/scala/org/apache/spark/sql/streaming/FlatMapGroupsInPandasWithStateSuite.scala
@@ -366,101 +366,91 @@ class FlatMapGroupsInPandasWithStateSuite extends StateStoreMetricsTest {
)
}
- Seq(true, false).map { ifUseDateTimeType =>
- test("applyInPandasWithState - streaming w/ event time timeout + watermark " +
- s"ifUseDateTimeType=$ifUseDateTimeType") {
- assume(shouldTestPandasUDFs)
-
- // timestamp_seconds assumes the base timezone is UTC. However, the provided function
- // localizes it. Therefore, this test assumes the timezone is in UTC
- withSQLConf(SQLConf.SESSION_LOCAL_TIMEZONE.key -> "UTC") {
- val timeoutMs = if (ifUseDateTimeType) {
- "datetime.datetime.fromtimestamp(timeout_timestamp_sec)"
- } else {
- "timeout_timestamp_sec * 1000"
- }
+ test("applyInPandasWithState - streaming w/ event time timeout + watermark") {
+ assume(shouldTestPandasUDFs)
- val pythonScript =
- s"""
- |import calendar
- |import os
- |import datetime
- |import pandas as pd
- |from pyspark.sql.types import StructType, StringType, StructField, IntegerType
- |
- |tpe = StructType([
- | StructField("key", StringType()),
- | StructField("maxEventTimeSec", IntegerType())])
- |
- |def func(key, pdf_iter, state):
- | assert state.getCurrentProcessingTimeMs() >= 0
- | assert state.getCurrentWatermarkMs() >= -1
- |
- | timeout_delay_sec = 5
- | if state.hasTimedOut:
- | state.remove()
- | yield pd.DataFrame({'key': [key[0]], 'maxEventTimeSec': [-1]})
- | else:
- | m = state.getOption
- | if m is None:
- | max_event_time_sec = 0
- | else:
- | max_event_time_sec = m[0]
- |
- | for pdf in pdf_iter:
- | pser = pdf.eventTime.apply(
- | lambda dt: (int(calendar.timegm(dt.utctimetuple()) + dt.microsecond)))
- | max_event_time_sec = int(max(pser.max(), max_event_time_sec))
- |
- | state.update((max_event_time_sec,))
- | timeout_timestamp_sec = max_event_time_sec + timeout_delay_sec
- | state.setTimeoutTimestamp($timeoutMs)
- | yield pd.DataFrame({'key': [key[0]],
- | 'maxEventTimeSec': [max_event_time_sec]})
- |""".stripMargin.format("")
- val pythonFunc = TestGroupedMapPandasUDFWithState(
- name = "pandas_grouped_map_with_state", pythonScript = pythonScript)
+ // timestamp_seconds assumes the base timezone is UTC. However, the provided function
+ // localizes it. Therefore, this test assumes the timezone is in UTC
+ withSQLConf(SQLConf.SESSION_LOCAL_TIMEZONE.key -> "UTC") {
+ val pythonScript =
+ """
+ |import calendar
+ |import os
+ |import datetime
+ |import pandas as pd
+ |from pyspark.sql.types import StructType, StringType, StructField, IntegerType
+ |
+ |tpe = StructType([
+ | StructField("key", StringType()),
+ | StructField("maxEventTimeSec", IntegerType())])
+ |
+ |def func(key, pdf_iter, state):
+ | assert state.getCurrentProcessingTimeMs() >= 0
+ | assert state.getCurrentWatermarkMs() >= -1
+ |
+ | timeout_delay_sec = 5
+ | if state.hasTimedOut:
+ | state.remove()
+ | yield pd.DataFrame({'key': [key[0]], 'maxEventTimeSec': [-1]})
+ | else:
+ | m = state.getOption
+ | if m is None:
+ | max_event_time_sec = 0
+ | else:
+ | max_event_time_sec = m[0]
+ |
+ | for pdf in pdf_iter:
+ | pser = pdf.eventTime.apply(
+ | lambda dt: (int(calendar.timegm(dt.utctimetuple()) + dt.microsecond)))
+ | max_event_time_sec = int(max(pser.max(), max_event_time_sec))
+ |
+ | state.update((max_event_time_sec,))
+ | timeout_timestamp_sec = max_event_time_sec + timeout_delay_sec
+ | state.setTimeoutTimestamp(timeout_timestamp_sec * 1000)
+ | yield pd.DataFrame({'key': [key[0]],
+ | 'maxEventTimeSec': [max_event_time_sec]})
+ |""".stripMargin
+ val pythonFunc = TestGroupedMapPandasUDFWithState(
+ name = "pandas_grouped_map_with_state", pythonScript = pythonScript)
- val inputData = MemoryStream[(String, Int)]
- val inputDataDF =
- inputData.toDF().select($"_1".as("key"), timestamp_seconds($"_2").as("eventTime"))
- val outputStructType = StructType(
- Seq(
- StructField("key", StringType),
- StructField("maxEventTimeSec", IntegerType)))
- val stateStructType = StructType(Seq(StructField("maxEventTimeSec", LongType)))
- val result =
- inputDataDF
- .withWatermark("eventTime", "10 seconds")
- .groupBy("key")
- .applyInPandasWithState(
- pythonFunc(inputDataDF("key"), inputDataDF("eventTime")).expr.asInstanceOf[PythonUDF],
- outputStructType,
- stateStructType,
- "Update",
- "EventTimeTimeout")
+ val inputData = MemoryStream[(String, Int)]
+ val inputDataDF =
+ inputData.toDF.select($"_1".as("key"), timestamp_seconds($"_2").as("eventTime"))
+ val outputStructType = StructType(
+ Seq(
+ StructField("key", StringType),
+ StructField("maxEventTimeSec", IntegerType)))
+ val stateStructType = StructType(Seq(StructField("maxEventTimeSec", LongType)))
+ val result =
+ inputDataDF
+ .withWatermark("eventTime", "10 seconds")
+ .groupBy("key")
+ .applyInPandasWithState(
+ pythonFunc(inputDataDF("key"), inputDataDF("eventTime")).expr.asInstanceOf[PythonUDF],
+ outputStructType,
+ stateStructType,
+ "Update",
+ "EventTimeTimeout")
- testStream(result, Update)(
- StartStream(),
+ testStream(result, Update)(
+ StartStream(),
- AddData(inputData, ("a", 11), ("a", 13), ("a", 15)),
- // Max event time = 15. Timeout timestamp for "a" = 15 + 5 = 20. Watermark = 15 - 10 = 5.
- CheckNewAnswer(("a", 15)), // Output = max event time of a
+ AddData(inputData, ("a", 11), ("a", 13), ("a", 15)),
+ // Max event time = 15. Timeout timestamp for "a" = 15 + 5 = 20. Watermark = 15 - 10 = 5.
+ CheckNewAnswer(("a", 15)), // Output = max event time of a
- AddData(inputData, ("a", 4)), // Add data older than watermark for "a"
- CheckNewAnswer(), // No output as data should get filtered by watermark
+ AddData(inputData, ("a", 4)), // Add data older than watermark for "a"
+ CheckNewAnswer(), // No output as data should get filtered by watermark
- AddData(inputData, ("a", 10)), // Add data newer than watermark for "a"
- CheckNewAnswer(("a", 15)), // Max event time is still the same
- // Timeout timestamp for "a" is still 20 as max event time for "a" is still 15.
- // Watermark is still 5 as max event time for all data is still 15.
+ AddData(inputData, ("a", 10)), // Add data newer than watermark for "a"
+ CheckNewAnswer(("a", 15)), // Max event time is still the same
+ // Timeout timestamp for "a" is still 20 as max event time for "a" is still 15.
+ // Watermark is still 5 as max event time for all data is still 15.
- AddData(inputData, ("b", 31)), // Add data newer than watermark for "b", not "a"
- // Watermark = 31 - 10 = 21, so "a" should be timed out as timeout timestamp for "a" is
- // 20.
- CheckNewAnswer(("a", -1), ("b", 31)) // State for "a" should timeout and emit -1
- )
- }
+ AddData(inputData, ("b", 31)), // Add data newer than watermark for "b", not "a"
+ // Watermark = 31 - 10 = 21, so "a" should be timed out as timeout timestamp for "a" is 20.
+ CheckNewAnswer(("a", -1), ("b", 31)) // State for "a" should timeout and emit -1
+ )
}
}
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@spark.apache.org
For additional commands, e-mail: commits-help@spark.apache.org