funrollloops commented on code in PR #56786:
URL: https://github.com/apache/spark/pull/56786#discussion_r3483453391
##########
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:
Thank you, good catch. I've moved the list of types into a frozenset that
will be initialized once to further speed this up.
--
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]
