You are viewing a plain text version of this content. The canonical link for it is here.
Posted to reviews@spark.apache.org by "HeartSaVioR (via GitHub)" <gi...@apache.org> on 2025/04/16 04:53:23 UTC
[PR] [SPARK-51814][SS] Introduce a new API `transformWithState` in PySpark [spark]
HeartSaVioR opened a new pull request, #50600:
URL: https://github.com/apache/spark/pull/50600
### What changes were proposed in this pull request?
This PR proposes to introduce a new API named `transformWithState` in PySpark, which uses PySpark Row type as the type of input and return type for user function. Note that this is not intended to replace the existing API `transformWithStateInPandas`, since users may rely on the feature of vectorization and also libraries which depend on Pandas.
The new API is quite same with existing `transformWithStateInPandas`, except the method name and the signature of user functions. The difference of signature of user functions are also quite straightforward - replace Pandas with Row.
There are several points on reviewing:
1. This PR tries to minimize the code diff via refactoring existing classes and reusing them. These classes are mostly renamed from Pandas to PySpark, because it's no longer used only for Pandas. Spark will pass user facing type as a parameter for internal execution classes to distinguish Pandas vs Row execution.
2. Same applies to config, hence this PR renames the existing config from `spark.sql.execution.arrow.transformWithStateInPandas.maxRecordsPerBatch` to `spark.sql.execution.arrow.transformWithStateInPySpark.maxRecordsPerBatch`. This PR retains the support of old name via `withAlternative` to not break backward compatibility.
3. From 1, `TransformWithStateInPandasExec` is renamed to `TransformWithStateInPySparkExec`. Since this affects both state operation metadata and state store reader, this PR retains the short name of physical node to `transformWithStateInPandasExec` when user facing type is set to Pandas.
4. Python test is majorly refactored to allow testing both Pandas and Row. Regardless of the type, the query is expected to produce the same output as long as we implement the stateful processor for the same behavior.
This PR does not address the support of Spark Connect - while it is easy and straightforward to support, we have to modify the proto which needs more eyes.
### Why are the changes needed?
We realized the overhead on converting Arrow RecordBatch to Pandas DataFrame is non-trivial, especially the case where cardinality is quite high (e.g. every data has its unique key). This case we see more than 10 times slower than the case of low cardinality. Using Row reduces a lot on the overhead of conversion, which addresses the huge gap of performance.
(We considered Arrow as a type since there are several pairs of Arrow <-> Pandas for existing PySpark API. We rejected the approach because we don't see a strong evidence that Arrow type of API is being widely used, and PySpark Row is lot easier to use.)
### Does this PR introduce _any_ user-facing change?
Yes, this PR introduces a new user facing API.
### How was this patch tested?
Tests are refactored to verify the new API.
### Was this patch authored or co-authored using generative AI tooling?
No.
--
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-51814][SS] Introduce a new API `transformWithState` in PySpark [spark]
Posted by "HeartSaVioR (via GitHub)" <gi...@apache.org>.
HeartSaVioR commented on code in PR #50600:
URL: https://github.com/apache/spark/pull/50600#discussion_r2049935483
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/internal/SQLConf.scala:
##########
@@ -3583,11 +3583,15 @@ object SQLConf {
"than zero and less than INT_MAX.")
.createWithDefaultString("256MB")
- val ARROW_TRANSFORM_WITH_STATE_IN_PANDAS_MAX_RECORDS_PER_BATCH =
- buildConf("spark.sql.execution.arrow.transformWithStateInPandas.maxRecordsPerBatch")
- .doc("When using TransformWithStateInPandas, limit the maximum number of state records " +
- "that can be written to a single ArrowRecordBatch in memory.")
+ val ARROW_TRANSFORM_WITH_STATE_IN_PYSPARK_MAX_RECORDS_PER_BATCH =
+ buildConf("spark.sql.execution.arrow.transformWithStateInPySpark.maxRecordsPerBatch")
Review Comment:
Yeah we had a discussion about this and we seem to miss it. I'll address altogether.
--
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-51814][SS] Introduce a new API `transformWithState` in PySpark [spark]
Posted by "anishshri-db (via GitHub)" <gi...@apache.org>.
anishshri-db commented on code in PR #50600:
URL: https://github.com/apache/spark/pull/50600#discussion_r2047902178
##########
python/pyspark/sql/tests/pandas/test_pandas_transform_with_state.py:
##########
@@ -49,31 +45,43 @@
pyarrow_requirement_message,
)
-if have_pandas:
- import pandas as pd
+from pyspark.sql.tests.pandas.helper.helper_pandas_transform_with_state import (
+ SimpleStatefulProcessorWithInitialStateFactory,
+ StatefulProcessorWithInitialStateTimersFactory,
+ StatefulProcessorWithListStateInitialStateFactory,
+ EventTimeStatefulProcessorFactory,
+ ProcTimeStatefulProcessorFactory,
+ SimpleStatefulProcessorFactory,
+ StatefulProcessorChainingOpsFactory,
+ SimpleTTLStatefulProcessorFactory,
+ TTLStatefulProcessorFactory,
+ InvalidSimpleStatefulProcessorFactory,
+ ListStateProcessorFactory,
+ ListStateLargeTTLProcessorFactory,
+ MapStateProcessorFactory,
+ MapStateLargeTTLProcessorFactory,
+ BasicProcessorFactory,
+ BasicProcessorNotNullableFactory,
+ AddFieldsProcessorFactory,
+ RemoveFieldsProcessorFactory,
+ ReorderedFieldsProcessorFactory,
+ UpcastProcessorFactory,
+ MinEventTimeStatefulProcessorFactory,
+)
-@unittest.skipIf(
- not have_pandas or not have_pyarrow or os.environ.get("PYTHON_GIL", "?") == "0",
- cast(
- str,
- pandas_requirement_message or pyarrow_requirement_message or "Not supported in no-GIL mode",
- ),
-)
-class TransformWithStateInPandasTestsMixin:
+class TransformWithStateTestsMixin:
@classmethod
- def conf(cls):
- cfg = SparkConf()
- cfg.set("spark.sql.shuffle.partitions", "5")
- cfg.set(
- "spark.sql.streaming.stateStore.providerClass",
- "org.apache.spark.sql.execution.streaming.state.RocksDBStateStoreProvider",
- )
- cfg.set("spark.sql.execution.arrow.transformWithStateInPandas.maxRecordsPerBatch", "2")
- cfg.set("spark.sql.session.timeZone", "UTC")
- # TODO SPARK-49046 this config is to stop query from FEB sink gracefully
- cfg.set("spark.sql.streaming.noDataMicroBatches.enabled", "false")
- return cfg
+ @abstractmethod
+ def use_pandas(cls) -> bool:
+ ...
+
+ @classmethod
+ def get_processor(cls, stateful_processor_factory) -> StatefulProcessor:
+ if cls.use_pandas():
+ return stateful_processor_factory.pandas()
+ else:
+ return stateful_processor_factory.row()
Review Comment:
Any tests needed specifically for the new interface ?
--
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-51814][SS] Introduce a new API `transformWithState` in PySpark [spark]
Posted by "anishshri-db (via GitHub)" <gi...@apache.org>.
anishshri-db commented on code in PR #50600:
URL: https://github.com/apache/spark/pull/50600#discussion_r2047861034
##########
python/pyspark/sql/pandas/serializers.py:
##########
@@ -1369,8 +1374,212 @@ def flatten_columns(cur_batch, col_name):
data_batches = generate_data_batches(_batches)
for k, g in groupby(data_batches, key=lambda x: x[0]):
- yield (TransformWithStateInPandasFuncMode.PROCESS_DATA, k, g)
+ yield (TransformWithStateInPySparkFuncMode.PROCESS_DATA, k, g)
+
+ yield (TransformWithStateInPySparkFuncMode.PROCESS_TIMER, None, None)
+
+ yield (TransformWithStateInPySparkFuncMode.COMPLETE, None, None)
+
+
+class TransformWithStateInPySparkRowSerializer(ArrowStreamUDFSerializer):
+ """
+ Serializer used by Python worker to evaluate UDF for
+ :meth:`pyspark.sql.GroupedData.transformWithState`.
+
+ Parameters
+ ----------
+ arrow_max_records_per_batch : int
+ Limit of the number of records that can be written to a single ArrowRecordBatch in memory.
+ """
+
+ def __init__(self, arrow_max_records_per_batch):
+ super(TransformWithStateInPySparkRowSerializer, self).__init__()
+ self.arrow_max_records_per_batch = arrow_max_records_per_batch
+ self.key_offsets = None
+
+ def load_stream(self, stream):
+ """
+ Read ArrowRecordBatches from stream, deserialize them to populate a list of data chunk, and
+ convert the data into a list of pandas.Series.
+
+ Please refer the doc of inner function `generate_data_batches` for more details how
+ this function works in overall.
+ """
+ from pyspark.sql.streaming.stateful_processor_util import (
+ TransformWithStateInPySparkFuncMode,
+ )
+ import itertools
+
+ def generate_data_batches(batches):
+ """
+ Deserialize ArrowRecordBatches and return a generator of Row.
+
+ The deserialization logic assumes that Arrow RecordBatches contain the data with the
+ ordering that data chunks for same grouping key will appear sequentially.
+
+ This function must avoid materializing multiple Arrow RecordBatches into memory at the
+ same time. And data chunks from the same grouping key should appear sequentially.
+ """
+ for batch in batches:
+ DataRow = Row(*(batch.column_names))
+
+ # This is supposed to be the same.
+ batch_key = tuple(batch[o][0].as_py() for o in self.key_offsets)
+ for row_idx in range(batch.num_rows):
+ row = DataRow(
+ *(batch.column(i)[row_idx].as_py() for i in range(batch.num_columns))
+ )
+ yield (batch_key, row)
+
+ _batches = super(ArrowStreamUDFSerializer, self).load_stream(stream)
+ data_batches = generate_data_batches(_batches)
+
+ for k, g in groupby(data_batches, key=lambda x: x[0]):
+ chained = itertools.chain(g)
+ chained_values = map(lambda x: x[1], chained)
+ yield (TransformWithStateInPySparkFuncMode.PROCESS_DATA, k, chained_values)
+
+ yield (TransformWithStateInPySparkFuncMode.PROCESS_TIMER, None, None)
+
+ yield (TransformWithStateInPySparkFuncMode.COMPLETE, None, None)
+
+ def dump_stream(self, iterator, stream):
+ """
+ Read through an iterator of (iterator of Row), serialize them to Arrow
+ RecordBatches, and write batches to stream.
+ """
+ import pyarrow as pa
+
+ def flatten_iterator():
+ # iterator: iter[list[(iter[Row], pdf_type)]]
+ for packed in iterator:
+ iter_row_with_type = packed[0]
+ iter_row = iter_row_with_type[0]
+ pdf_type = iter_row_with_type[1]
+
+ rows_as_dict = []
+ for row in iter_row:
+ row_as_dict = row.asDict(True)
+ rows_as_dict.append(row_as_dict)
+
+ pdf_schema = pa.schema(pdf_type.fields)
+ record_batch = pa.RecordBatch.from_pylist(rows_as_dict, schema=pdf_schema)
+
+ yield (record_batch, pdf_type)
+
+ return ArrowStreamUDFSerializer.dump_stream(self, flatten_iterator(), stream)
+
+
+class TransformWithStateInPySparkRowInitStateSerializer(TransformWithStateInPySparkRowSerializer):
+ """
+ Serializer used by Python worker to evaluate UDF for
+ :meth:`pyspark.sql.GroupedData.transformWithStateInPySparkRowInitStateSerializer`.
+ Parameters
+ ----------
+ Same as input parameters in TransformWithStateInPySparkRowSerializer.
+ """
+
+ def __init__(self, arrow_max_records_per_batch):
+ super(TransformWithStateInPySparkRowInitStateSerializer, self).__init__(
+ arrow_max_records_per_batch
+ )
+ self.init_key_offsets = None
+
+ def load_stream(self, stream):
+ import itertools
+ import pyarrow as pa
+ from pyspark.sql.streaming.stateful_processor_util import (
+ TransformWithStateInPySparkFuncMode,
+ )
+
+ def generate_data_batches(batches):
+ """
+ Deserialize ArrowRecordBatches and return a generator of Row.
+ The deserialization logic assumes that Arrow RecordBatches contain the data with the
+ ordering that data chunks for same grouping key will appear sequentially.
+ See `TransformWithStateInPySparkPythonInitialStateRunner` for arrow batch schema sent
+ from JVM.
+ This function flatten the columns of input rows and initial state rows and feed them
Review Comment:
nit: `flattens the columns of`
##########
python/pyspark/sql/pandas/serializers.py:
##########
@@ -1369,8 +1374,212 @@ def flatten_columns(cur_batch, col_name):
data_batches = generate_data_batches(_batches)
for k, g in groupby(data_batches, key=lambda x: x[0]):
- yield (TransformWithStateInPandasFuncMode.PROCESS_DATA, k, g)
+ yield (TransformWithStateInPySparkFuncMode.PROCESS_DATA, k, g)
+
+ yield (TransformWithStateInPySparkFuncMode.PROCESS_TIMER, None, None)
+
+ yield (TransformWithStateInPySparkFuncMode.COMPLETE, None, None)
+
+
+class TransformWithStateInPySparkRowSerializer(ArrowStreamUDFSerializer):
+ """
+ Serializer used by Python worker to evaluate UDF for
+ :meth:`pyspark.sql.GroupedData.transformWithState`.
+
+ Parameters
+ ----------
+ arrow_max_records_per_batch : int
+ Limit of the number of records that can be written to a single ArrowRecordBatch in memory.
+ """
+
+ def __init__(self, arrow_max_records_per_batch):
+ super(TransformWithStateInPySparkRowSerializer, self).__init__()
+ self.arrow_max_records_per_batch = arrow_max_records_per_batch
+ self.key_offsets = None
+
+ def load_stream(self, stream):
+ """
+ Read ArrowRecordBatches from stream, deserialize them to populate a list of data chunk, and
+ convert the data into a list of pandas.Series.
Review Comment:
There is no `Pandas` here any more right ?
##########
python/pyspark/sql/streaming/stateful_processor.py:
##########
@@ -380,25 +388,29 @@ def init(self, handle: StatefulProcessorHandle) -> None:
def handleInputRows(
self,
key: Any,
- rows: Iterator["PandasDataFrameLike"],
+ rows: Union[Iterator["PandasDataFrameLike"], Iterator[Row]],
timerValues: TimerValues,
- ) -> Iterator["PandasDataFrameLike"]:
+ ) -> Union[Iterator["PandasDataFrameLike"], Iterator[Row]]:
"""
Function that will allow users to interact with input data rows along with the grouping key.
- It should take parameters (key, Iterator[`pandas.DataFrame`]) and return another
- Iterator[`pandas.DataFrame`]. For each group, all columns are passed together as
- `pandas.DataFrame` to the function, and the returned `pandas.DataFrame` across all
- invocations are combined as a :class:`DataFrame`. Note that the function should not make a
- guess of the number of elements in the iterator. To process all data, the `handleInputRows`
- function needs to iterate all elements and process them. On the other hand, the
- `handleInputRows` function is not strictly required to iterate through all elements in the
- iterator if it intends to read a part of data.
+
+ The type of input data and return are different by which method is called.
Review Comment:
nit: same/similar as above
##########
python/pyspark/sql/streaming/stateful_processor.py:
##########
@@ -359,6 +359,14 @@ class StatefulProcessor(ABC):
Class that represents the arbitrary stateful logic that needs to be provided by the user to
perform stateful manipulations on keyed streams.
+ NOTE: The type of input data and return are different by which method is called, following:
Review Comment:
nit: `type of input data and return type are different based on which method is called, such as`
##########
python/pyspark/sql/pandas/serializers.py:
##########
@@ -1369,8 +1374,212 @@ def flatten_columns(cur_batch, col_name):
data_batches = generate_data_batches(_batches)
for k, g in groupby(data_batches, key=lambda x: x[0]):
- yield (TransformWithStateInPandasFuncMode.PROCESS_DATA, k, g)
+ yield (TransformWithStateInPySparkFuncMode.PROCESS_DATA, k, g)
+
+ yield (TransformWithStateInPySparkFuncMode.PROCESS_TIMER, None, None)
+
+ yield (TransformWithStateInPySparkFuncMode.COMPLETE, None, None)
+
+
+class TransformWithStateInPySparkRowSerializer(ArrowStreamUDFSerializer):
+ """
+ Serializer used by Python worker to evaluate UDF for
+ :meth:`pyspark.sql.GroupedData.transformWithState`.
+
+ Parameters
+ ----------
+ arrow_max_records_per_batch : int
+ Limit of the number of records that can be written to a single ArrowRecordBatch in memory.
+ """
+
+ def __init__(self, arrow_max_records_per_batch):
+ super(TransformWithStateInPySparkRowSerializer, self).__init__()
+ self.arrow_max_records_per_batch = arrow_max_records_per_batch
+ self.key_offsets = None
+
+ def load_stream(self, stream):
+ """
+ Read ArrowRecordBatches from stream, deserialize them to populate a list of data chunk, and
Review Comment:
nit: `list of data chunks`
--
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-51814][SS] Introduce a new API `transformWithState` in PySpark [spark]
Posted by "anishshri-db (via GitHub)" <gi...@apache.org>.
anishshri-db commented on code in PR #50600:
URL: https://github.com/apache/spark/pull/50600#discussion_r2047890734
##########
python/pyspark/sql/tests/connect/pandas/test_parity_pandas_transform_with_state.py:
##########
@@ -53,6 +55,37 @@ def test_schema_evolution_scenarios(self):
pass
+# TODO: need to unblock once we implement transformWithState in Python Spark Connect
Review Comment:
Should we remove till then ?
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/state/StateDataSource.scala:
##########
@@ -55,7 +55,11 @@ class StateDataSource extends TableProvider with DataSourceRegister with Logging
// Seq of operator names who uses state schema v3 and TWS related options.
// This Seq was used in checks before reading state schema files.
- private val twsShortNameSeq = Seq("transformWithStateExec", "transformWithStateInPandasExec")
+ private val twsShortNameSeq = Seq(
+ "transformWithStateExec",
+ "transformWithStateInPandasExec",
+ "transformWithStateInPySparkExec"
Review Comment:
Can we add tests for the state source reader with the new interface also ?
--
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-51814][SS] Introduce a new API `transformWithState` in PySpark [spark]
Posted by "bogao007 (via GitHub)" <gi...@apache.org>.
bogao007 commented on code in PR #50600:
URL: https://github.com/apache/spark/pull/50600#discussion_r2049705985
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/internal/SQLConf.scala:
##########
@@ -3583,11 +3583,15 @@ object SQLConf {
"than zero and less than INT_MAX.")
.createWithDefaultString("256MB")
- val ARROW_TRANSFORM_WITH_STATE_IN_PANDAS_MAX_RECORDS_PER_BATCH =
- buildConf("spark.sql.execution.arrow.transformWithStateInPandas.maxRecordsPerBatch")
- .doc("When using TransformWithStateInPandas, limit the maximum number of state records " +
- "that can be written to a single ArrowRecordBatch in memory.")
+ val ARROW_TRANSFORM_WITH_STATE_IN_PYSPARK_MAX_RECORDS_PER_BATCH =
+ buildConf("spark.sql.execution.arrow.transformWithStateInPySpark.maxRecordsPerBatch")
Review Comment:
Since we have a chance to update this config name, should we use `maxStateRecordsPerBatch` to make it more specific that this config is to control the state record size?
##########
python/pyspark/sql/tests/connect/pandas/test_parity_pandas_transform_with_state.py:
##########
@@ -17,14 +17,14 @@
import unittest
from pyspark.sql.tests.pandas.test_pandas_transform_with_state import (
- TransformWithStateInPandasTestsMixin,
+ TransformWithStateInPandasTestsMixIn,
Review Comment:
Nit: did you accidentally update it to be `MixIn` from `Mixin`?
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/state/StateDataSource.scala:
##########
@@ -55,7 +55,11 @@ class StateDataSource extends TableProvider with DataSourceRegister with Logging
// Seq of operator names who uses state schema v3 and TWS related options.
// This Seq was used in checks before reading state schema files.
- private val twsShortNameSeq = Seq("transformWithStateExec", "transformWithStateInPandasExec")
+ private val twsShortNameSeq = Seq(
+ "transformWithStateExec",
+ "transformWithStateInPandasExec",
+ "transformWithStateInPySparkExec"
Review Comment:
Do we keep `transformWithStateInPandasExec` because of compatibility reasons?
--
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-51814][SS] Introduce a new API `transformWithState` in PySpark [spark]
Posted by "HeartSaVioR (via GitHub)" <gi...@apache.org>.
HeartSaVioR commented on PR #50600:
URL: https://github.com/apache/spark/pull/50600#issuecomment-2808295724
Almost 2500 lines of changes are bound to the test (both refactor and new addition). The change on the main code is not that significant.
--
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-51814][SS] Introduce a new API `transformWithState` in PySpark [spark]
Posted by "HeartSaVioR (via GitHub)" <gi...@apache.org>.
HeartSaVioR commented on PR #50600:
URL: https://github.com/apache/spark/pull/50600#issuecomment-2808296674
cc. @anishshri-db @bogao007 @jingz-db
Please take a 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-51814][SS] Introduce a new API `transformWithState` in PySpark [spark]
Posted by "HeartSaVioR (via GitHub)" <gi...@apache.org>.
HeartSaVioR commented on code in PR #50600:
URL: https://github.com/apache/spark/pull/50600#discussion_r2048051579
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/python/streaming/TransformWithStateInPySparkExec.scala:
##########
@@ -494,7 +521,7 @@ object TransformWithStateInPandasExec {
stateStoreCkptIds = None
)
- new TransformWithStateInPandasExec(
+ new TransformWithStateInPySparkExec(
Review Comment:
Yes, but we retain the "shortName" as it is for Pandas path, so wouldn't have a compatibility issue.
--
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-51814][SS] Introduce a new API `transformWithState` in PySpark [spark]
Posted by "HeartSaVioR (via GitHub)" <gi...@apache.org>.
HeartSaVioR commented on code in PR #50600:
URL: https://github.com/apache/spark/pull/50600#discussion_r2048041661
##########
python/pyspark/sql/pandas/serializers.py:
##########
@@ -1369,8 +1374,212 @@ def flatten_columns(cur_batch, col_name):
data_batches = generate_data_batches(_batches)
for k, g in groupby(data_batches, key=lambda x: x[0]):
- yield (TransformWithStateInPandasFuncMode.PROCESS_DATA, k, g)
+ yield (TransformWithStateInPySparkFuncMode.PROCESS_DATA, k, g)
+
+ yield (TransformWithStateInPySparkFuncMode.PROCESS_TIMER, None, None)
+
+ yield (TransformWithStateInPySparkFuncMode.COMPLETE, None, None)
+
+
+class TransformWithStateInPySparkRowSerializer(ArrowStreamUDFSerializer):
+ """
+ Serializer used by Python worker to evaluate UDF for
+ :meth:`pyspark.sql.GroupedData.transformWithState`.
+
+ Parameters
+ ----------
+ arrow_max_records_per_batch : int
+ Limit of the number of records that can be written to a single ArrowRecordBatch in memory.
+ """
+
+ def __init__(self, arrow_max_records_per_batch):
+ super(TransformWithStateInPySparkRowSerializer, self).__init__()
+ self.arrow_max_records_per_batch = arrow_max_records_per_batch
+ self.key_offsets = None
+
+ def load_stream(self, stream):
+ """
+ Read ArrowRecordBatches from stream, deserialize them to populate a list of data chunk, and
+ convert the data into a list of pandas.Series.
Review Comment:
Yeah, nice finding!
--
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-51814][SS] Introduce a new API `transformWithState` in PySpark [spark]
Posted by "HeartSaVioR (via GitHub)" <gi...@apache.org>.
HeartSaVioR commented on code in PR #50600:
URL: https://github.com/apache/spark/pull/50600#discussion_r2048050612
##########
sql/core/src/main/scala/org/apache/spark/sql/classic/RelationalGroupedDataset.scala:
##########
@@ -479,6 +505,24 @@ class RelationalGroupedDataset protected[sql](
timeModeStr: String,
initialState: RelationalGroupedDataset,
eventTimeColumnName: String): DataFrame = {
+ _transformWithStateInPySpark(
+ func,
+ outputStructType,
+ outputModeStr,
+ timeModeStr,
+ initialState,
+ eventTimeColumnName,
+ TransformWithStateInPySpark.UserFacingDataType.PANDAS)
+ }
+
+ private def _transformWithStateInPySpark(
+ func: Column,
+ outputStructType: StructType,
+ outputModeStr: String,
+ timeModeStr: String,
+ initialState: RelationalGroupedDataset,
+ eventTimeColumnName: String,
+ userFacingDataType: TransformWithStateInPySpark.UserFacingDataType.Value): DataFrame = {
Review Comment:
I'm not sure this is complicated - what else you want to propose?
--
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-51814][SS][PYTHON] Introduce a new API `transformWithState` in PySpark [spark]
Posted by "HeartSaVioR (via GitHub)" <gi...@apache.org>.
HeartSaVioR commented on PR #50600:
URL: https://github.com/apache/spark/pull/50600#issuecomment-2814416058
@cloud-fan @HyukjinKwon Could you please help talking a 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-51814][SS] Introduce a new API `transformWithState` in PySpark [spark]
Posted by "HeartSaVioR (via GitHub)" <gi...@apache.org>.
HeartSaVioR commented on code in PR #50600:
URL: https://github.com/apache/spark/pull/50600#discussion_r2049936823
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/state/StateDataSource.scala:
##########
@@ -55,7 +55,11 @@ class StateDataSource extends TableProvider with DataSourceRegister with Logging
// Seq of operator names who uses state schema v3 and TWS related options.
// This Seq was used in checks before reading state schema files.
- private val twsShortNameSeq = Seq("transformWithStateExec", "transformWithStateInPandasExec")
+ private val twsShortNameSeq = Seq(
+ "transformWithStateExec",
+ "transformWithStateInPandasExec",
+ "transformWithStateInPySparkExec"
Review Comment:
Yes; shortName is used in operator metadata.
--
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-51814][SS] Introduce a new API `transformWithState` in PySpark [spark]
Posted by "anishshri-db (via GitHub)" <gi...@apache.org>.
anishshri-db commented on code in PR #50600:
URL: https://github.com/apache/spark/pull/50600#discussion_r2048157541
##########
python/pyspark/sql/streaming/stateful_processor.py:
##########
@@ -359,6 +359,14 @@ class StatefulProcessor(ABC):
Class that represents the arbitrary stateful logic that needs to be provided by the user to
perform stateful manipulations on keyed streams.
+ NOTE: The type of input data and return are different by which method is called, following:
Review Comment:
Did we update these or you prefer to retain what you have ?
--
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-51814][SS] Introduce a new API `transformWithState` in PySpark [spark]
Posted by "HeartSaVioR (via GitHub)" <gi...@apache.org>.
HeartSaVioR commented on code in PR #50600:
URL: https://github.com/apache/spark/pull/50600#discussion_r2048049596
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/UnsupportedOperationChecker.scala:
##########
@@ -105,7 +105,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:
The class name of physical node has changed, so no need to do that.
--
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-51814][SS] Introduce a new API `transformWithState` in PySpark [spark]
Posted by "HeartSaVioR (via GitHub)" <gi...@apache.org>.
HeartSaVioR commented on code in PR #50600:
URL: https://github.com/apache/spark/pull/50600#discussion_r2049940576
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/internal/SQLConf.scala:
##########
@@ -3583,11 +3583,15 @@ object SQLConf {
"than zero and less than INT_MAX.")
.createWithDefaultString("256MB")
- val ARROW_TRANSFORM_WITH_STATE_IN_PANDAS_MAX_RECORDS_PER_BATCH =
- buildConf("spark.sql.execution.arrow.transformWithStateInPandas.maxRecordsPerBatch")
- .doc("When using TransformWithStateInPandas, limit the maximum number of state records " +
- "that can be written to a single ArrowRecordBatch in memory.")
+ val ARROW_TRANSFORM_WITH_STATE_IN_PYSPARK_MAX_RECORDS_PER_BATCH =
+ buildConf("spark.sql.execution.arrow.transformWithStateInPySpark.maxRecordsPerBatch")
Review Comment:
Done
--
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-51814][SS] Introduce a new API `transformWithState` in PySpark [spark]
Posted by "anishshri-db (via GitHub)" <gi...@apache.org>.
anishshri-db commented on code in PR #50600:
URL: https://github.com/apache/spark/pull/50600#discussion_r2047856307
##########
python/pyspark/sql/pandas/group_ops.py:
##########
@@ -481,10 +481,162 @@ def transformWithStateInPandas(
-----
This function requires a full shuffle.
"""
+ return self.__transformWithState(
+ statefulProcessor,
+ outputStructType,
+ outputMode,
+ timeMode,
+ True,
+ initialState,
+ eventTimeColumnName,
+ )
+
+ def transformWithState(
+ self,
+ statefulProcessor: StatefulProcessor,
+ outputStructType: Union[StructType, str],
+ outputMode: str,
+ timeMode: str,
+ initialState: Optional["GroupedData"] = None,
+ eventTimeColumnName: str = "",
+ ) -> "DataFrame":
+ """
+ Invokes methods defined in the stateful processor used in arbitrary state API v2. It
+ requires protobuf and pyarrow as dependencies to process input/state data. We allow
+ the user to act on per-group set of input rows along with keyed state and the user
+ can choose to output/return 0 or more rows.
+
+ For a streaming dataframe, we will repeatedly invoke the interface methods for new rows
+ in each trigger and the user's state/state variables will be stored persistently across
+ invocations.
+
+ The `statefulProcessor` should be a Python class that implements the interface defined in
+ :class:`StatefulProcessor`.
+
+ The `outputStructType` should be a :class:`StructType` describing the schema of all
+ elements in the returned value, `Row`. The column labels of all elements in
+ returned `Row` must either match the field names in the defined schema.
+
+ The number of `Row` in the iterator in both the input and output can be arbitrary.
Review Comment:
nit: `The number of `Row`s` ?
--
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-51814][SS] Introduce a new API `transformWithState` in PySpark [spark]
Posted by "anishshri-db (via GitHub)" <gi...@apache.org>.
anishshri-db commented on PR #50600:
URL: https://github.com/apache/spark/pull/50600#issuecomment-2810987993
@HeartSaVioR - CI seems like its failing - https://github.com/apache/spark/pull/50600/checks?check_run_id=40628849458 ?
Do you think its related ?
--
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-51814][SS] Introduce a new API `transformWithState` in PySpark [spark]
Posted by "HeartSaVioR (via GitHub)" <gi...@apache.org>.
HeartSaVioR commented on code in PR #50600:
URL: https://github.com/apache/spark/pull/50600#discussion_r2048049754
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/plans/logical/pythonLogicalOperators.scala:
##########
@@ -178,16 +178,18 @@ case class FlatMapGroupsInPandasWithState(
* @param outputAttrs used to define the output rows
* @param outputMode defines the output mode for the statefulProcessor
* @param timeMode the time mode semantics of the stateful processor for timers and TTL.
+ * // FIXME: document `userFacingDataType`
Review Comment:
Ouch that's missed spot.
--
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-51814][SS] Introduce a new API `transformWithState` in PySpark [spark]
Posted by "anishshri-db (via GitHub)" <gi...@apache.org>.
anishshri-db commented on code in PR #50600:
URL: https://github.com/apache/spark/pull/50600#discussion_r2048164106
##########
sql/core/src/main/scala/org/apache/spark/sql/classic/RelationalGroupedDataset.scala:
##########
@@ -479,6 +505,24 @@ class RelationalGroupedDataset protected[sql](
timeModeStr: String,
initialState: RelationalGroupedDataset,
eventTimeColumnName: String): DataFrame = {
+ _transformWithStateInPySpark(
+ func,
+ outputStructType,
+ outputModeStr,
+ timeModeStr,
+ initialState,
+ eventTimeColumnName,
+ TransformWithStateInPySpark.UserFacingDataType.PANDAS)
+ }
+
+ private def _transformWithStateInPySpark(
+ func: Column,
+ outputStructType: StructType,
+ outputModeStr: String,
+ timeModeStr: String,
+ initialState: RelationalGroupedDataset,
+ eventTimeColumnName: String,
+ userFacingDataType: TransformWithStateInPySpark.UserFacingDataType.Value): DataFrame = {
Review Comment:
Oh nvm - this is not directly exposed to users. Name is different between pandas and non-pandas anyway - so should be fine
--
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-51814][SS] Introduce a new API `transformWithState` in PySpark [spark]
Posted by "anishshri-db (via GitHub)" <gi...@apache.org>.
anishshri-db commented on code in PR #50600:
URL: https://github.com/apache/spark/pull/50600#discussion_r2048158611
##########
python/pyspark/sql/tests/connect/pandas/test_parity_pandas_transform_with_state.py:
##########
@@ -53,6 +55,8 @@ def test_schema_evolution_scenarios(self):
pass
+# TODO: Need to copy the parity test when we implement transformWithState in Python Spark Connect
Review Comment:
Should we also add the SPARK JIRA ticket for this support ?
--
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-51814][SS] Introduce a new API `transformWithState` in PySpark [spark]
Posted by "anishshri-db (via GitHub)" <gi...@apache.org>.
anishshri-db commented on code in PR #50600:
URL: https://github.com/apache/spark/pull/50600#discussion_r2047870718
##########
sql/core/src/main/scala/org/apache/spark/sql/classic/RelationalGroupedDataset.scala:
##########
@@ -479,6 +505,24 @@ class RelationalGroupedDataset protected[sql](
timeModeStr: String,
initialState: RelationalGroupedDataset,
eventTimeColumnName: String): DataFrame = {
+ _transformWithStateInPySpark(
+ func,
+ outputStructType,
+ outputModeStr,
+ timeModeStr,
+ initialState,
+ eventTimeColumnName,
+ TransformWithStateInPySpark.UserFacingDataType.PANDAS)
+ }
+
+ private def _transformWithStateInPySpark(
+ func: Column,
+ outputStructType: StructType,
+ outputModeStr: String,
+ timeModeStr: String,
+ initialState: RelationalGroupedDataset,
+ eventTimeColumnName: String,
+ userFacingDataType: TransformWithStateInPySpark.UserFacingDataType.Value): DataFrame = {
Review Comment:
Should we use a simpler name here maybe ? instead of `userFacingDataType` ?
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/plans/logical/pythonLogicalOperators.scala:
##########
@@ -178,16 +178,18 @@ case class FlatMapGroupsInPandasWithState(
* @param outputAttrs used to define the output rows
* @param outputMode defines the output mode for the statefulProcessor
* @param timeMode the time mode semantics of the stateful processor for timers and TTL.
+ * // FIXME: document `userFacingDataType`
Review Comment:
Lets add it in this PR itself ?
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/state/StateDataSource.scala:
##########
@@ -55,7 +55,11 @@ class StateDataSource extends TableProvider with DataSourceRegister with Logging
// Seq of operator names who uses state schema v3 and TWS related options.
// This Seq was used in checks before reading state schema files.
- private val twsShortNameSeq = Seq("transformWithStateExec", "transformWithStateInPandasExec")
+ private val twsShortNameSeq = Seq(
+ "transformWithStateExec",
+ "transformWithStateInPandasExec",
+ "transformWithStateInPySparkExec"
Review Comment:
Will this work as it is or we need to do more work to integrate with state data source reader ?
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/python/streaming/TransformWithStateInPySparkExec.scala:
##########
@@ -494,7 +521,7 @@ object TransformWithStateInPandasExec {
stateStoreCkptIds = None
)
- new TransformWithStateInPandasExec(
+ new TransformWithStateInPySparkExec(
Review Comment:
Are we changing the name of the physical operator also for the current pandas tws operator ?
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/UnsupportedOperationChecker.scala:
##########
@@ -105,7 +105,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:
Don't we need to retain this also ?
--
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-51814][SS] Introduce a new API `transformWithState` in PySpark [spark]
Posted by "HeartSaVioR (via GitHub)" <gi...@apache.org>.
HeartSaVioR commented on code in PR #50600:
URL: https://github.com/apache/spark/pull/50600#discussion_r2048052017
##########
python/pyspark/sql/tests/connect/pandas/test_parity_pandas_transform_with_state.py:
##########
@@ -53,6 +55,37 @@ def test_schema_evolution_scenarios(self):
pass
+# TODO: need to unblock once we implement transformWithState in Python Spark Connect
Review Comment:
Either is fine. I'll just leave TODO comment with removing codes.
--
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-51814][SS] Introduce a new API `transformWithState` in PySpark [spark]
Posted by "HeartSaVioR (via GitHub)" <gi...@apache.org>.
HeartSaVioR commented on code in PR #50600:
URL: https://github.com/apache/spark/pull/50600#discussion_r2048052841
##########
python/pyspark/sql/tests/pandas/test_pandas_transform_with_state.py:
##########
@@ -49,31 +45,43 @@
pyarrow_requirement_message,
)
-if have_pandas:
- import pandas as pd
+from pyspark.sql.tests.pandas.helper.helper_pandas_transform_with_state import (
+ SimpleStatefulProcessorWithInitialStateFactory,
+ StatefulProcessorWithInitialStateTimersFactory,
+ StatefulProcessorWithListStateInitialStateFactory,
+ EventTimeStatefulProcessorFactory,
+ ProcTimeStatefulProcessorFactory,
+ SimpleStatefulProcessorFactory,
+ StatefulProcessorChainingOpsFactory,
+ SimpleTTLStatefulProcessorFactory,
+ TTLStatefulProcessorFactory,
+ InvalidSimpleStatefulProcessorFactory,
+ ListStateProcessorFactory,
+ ListStateLargeTTLProcessorFactory,
+ MapStateProcessorFactory,
+ MapStateLargeTTLProcessorFactory,
+ BasicProcessorFactory,
+ BasicProcessorNotNullableFactory,
+ AddFieldsProcessorFactory,
+ RemoveFieldsProcessorFactory,
+ ReorderedFieldsProcessorFactory,
+ UpcastProcessorFactory,
+ MinEventTimeStatefulProcessorFactory,
+)
-@unittest.skipIf(
- not have_pandas or not have_pyarrow or os.environ.get("PYTHON_GIL", "?") == "0",
- cast(
- str,
- pandas_requirement_message or pyarrow_requirement_message or "Not supported in no-GIL mode",
- ),
-)
-class TransformWithStateInPandasTestsMixin:
+class TransformWithStateTestsMixin:
@classmethod
- def conf(cls):
- cfg = SparkConf()
- cfg.set("spark.sql.shuffle.partitions", "5")
- cfg.set(
- "spark.sql.streaming.stateStore.providerClass",
- "org.apache.spark.sql.execution.streaming.state.RocksDBStateStoreProvider",
- )
- cfg.set("spark.sql.execution.arrow.transformWithStateInPandas.maxRecordsPerBatch", "2")
- cfg.set("spark.sql.session.timeZone", "UTC")
- # TODO SPARK-49046 this config is to stop query from FEB sink gracefully
- cfg.set("spark.sql.streaming.noDataMicroBatches.enabled", "false")
- return cfg
+ @abstractmethod
+ def use_pandas(cls) -> bool:
+ ...
+
+ @classmethod
+ def get_processor(cls, stateful_processor_factory) -> StatefulProcessor:
+ if cls.use_pandas():
+ return stateful_processor_factory.pandas()
+ else:
+ return stateful_processor_factory.row()
Review Comment:
No, the only difference from existing API is user function, hence I think this should be OK as long as we ensure every existing tests to work with new API 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-51814][SS] Introduce a new API `transformWithState` in PySpark [spark]
Posted by "HeartSaVioR (via GitHub)" <gi...@apache.org>.
HeartSaVioR commented on code in PR #50600:
URL: https://github.com/apache/spark/pull/50600#discussion_r2048051205
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/state/StateDataSource.scala:
##########
@@ -55,7 +55,11 @@ class StateDataSource extends TableProvider with DataSourceRegister with Logging
// Seq of operator names who uses state schema v3 and TWS related options.
// This Seq was used in checks before reading state schema files.
- private val twsShortNameSeq = Seq("transformWithStateExec", "transformWithStateInPandasExec")
+ private val twsShortNameSeq = Seq(
+ "transformWithStateExec",
+ "transformWithStateInPandasExec",
+ "transformWithStateInPySparkExec"
Review Comment:
This PR enables existing tests for transformWithStateInPandas to run with transformWithState as well, which includes tests with state reader.
--
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-51814][SS][PYTHON] Introduce a new API `transformWithState` in PySpark [spark]
Posted by "HeartSaVioR (via GitHub)" <gi...@apache.org>.
HeartSaVioR closed pull request #50600: [SPARK-51814][SS][PYTHON] Introduce a new API `transformWithState` in PySpark
URL: https://github.com/apache/spark/pull/50600
--
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-51814][SS][PYTHON] Introduce a new API `transformWithState` in PySpark [spark]
Posted by "HeartSaVioR (via GitHub)" <gi...@apache.org>.
HeartSaVioR commented on PR #50600:
URL: https://github.com/apache/spark/pull/50600#issuecomment-2814917633
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-51814][SS] Introduce a new API `transformWithState` in PySpark [spark]
Posted by "HeartSaVioR (via GitHub)" <gi...@apache.org>.
HeartSaVioR commented on code in PR #50600:
URL: https://github.com/apache/spark/pull/50600#discussion_r2049935647
##########
python/pyspark/sql/tests/connect/pandas/test_parity_pandas_transform_with_state.py:
##########
@@ -17,14 +17,14 @@
import unittest
from pyspark.sql.tests.pandas.test_pandas_transform_with_state import (
- TransformWithStateInPandasTestsMixin,
+ TransformWithStateInPandasTestsMixIn,
Review Comment:
Didn't know I have made a change. Maybe I'll just revert the name.
--
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