HyukjinKwon commented on code in PR #56786:
URL: https://github.com/apache/spark/pull/56786#discussion_r3478098937
##########
python/pyspark/sql/streaming/stateful_processor_api_client.py:
##########
@@ -27,12 +29,43 @@
Row,
)
from pyspark.sql.pandas.types import convert_pandas_using_numpy_type
-from pyspark.serializers import CPickleSerializer
+from pyspark.serializers import PickleSerializer
from pyspark.errors import PySparkRuntimeError
import uuid
__all__ = ["StatefulProcessorApiClient", "StatefulProcessorHandleState"]
+try:
+ import numpy as np
+
+ has_numpy = True
+
+ def _normalize_state_value(v: Any) -> Any:
+ # Fast path for common scalar values.
+ if isinstance(v, (bool, int, float, str, bytes, datetime, NoneType)):
Review Comment:
This fast-path runs before the `np.generic` / `to_pydatetime` branches, so
it short-circuits subclasses of the listed base types. `np.float64` is a
subclass of `float` and `pandas.Timestamp` is a subclass of `datetime`, so both
now return **unconverted** here, where the old code normalized them to Python
`float`/`datetime` via `.tolist()`/`.to_pydatetime()`. (`np.int64`/`np.bool_`
are *not* subclasses of `int`/`bool`, so those are unaffected.)
Use exact-type checks so subclasses fall through to the conversion branches:
```suggestion
if type(v) in (bool, int, float, str, bytes, datetime, type(None)):
```
##########
python/pyspark/sql/streaming/stateful_processor_api_client.py:
##########
@@ -74,7 +107,7 @@ def __init__(
else:
self.handle_state = StatefulProcessorHandleState.CREATED
self.utf8_deserializer = UTF8Deserializer()
- self.pickleSer = CPickleSerializer()
+ self.pickleSer = PickleSerializer()
Review Comment:
This swaps the serializer, not just renames it. `CPickleSerializer` resolves
to `CloudPickleSerializer` by default — it only aliases to plain
`PickleSerializer` when `PYSPARK_ENABLE_NAMEDTUPLE_PATCH=1`
(`pyspark/serializers.py:459-462`). So hardcoding `PickleSerializer()` drops
cloudpickle's ability to serialize closures, lambdas, and locally-defined
classes that can appear in user state values.
If this is intentional (relying on the new namedtuple/Row normalization,
with plain pickle being faster), please call it out in the description and add
a test covering a state value that previously required cloudpickle. As written
it's an unexplained capability change in an "optimizations" PR.
##########
python/pyspark/sql/streaming/stateful_processor_api_client.py:
##########
@@ -14,11 +14,13 @@
# See the License for the specific language governing permissions and
# limitations under the License.
#
+from datetime import datetime
from enum import Enum
import json
import os
import socket
-from typing import IO, Any, Dict, List, Union, Optional, Tuple, Iterator, cast
+from types import NoneType
+from typing import IO, Any, Dict, List, Union, Optional, Tuple, Iterator,
cast, TYPE_CHECKING
Review Comment:
`TYPE_CHECKING` is imported but not used anywhere in the file — flake8 F401
will fail CI. Drop it (or add the intended `if TYPE_CHECKING:` block):
```suggestion
from typing import IO, Any, Dict, List, Union, Optional, Tuple, Iterator,
cast
```
--
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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
