This is an automated email from the ASF dual-hosted git repository.
HyukjinKwon pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/spark.git
The following commit(s) were added to refs/heads/master by this push:
new e4c894ad615e [SPARK-57760][SS][PYTHON][FOLLOWUP] Avoid importing numpy
at module level in stateful_processor_api_client
e4c894ad615e is described below
commit e4c894ad615e41bf9d0a3c26a48d0e9b99ced611
Author: Kousuke Saruta <[email protected]>
AuthorDate: Fri Jul 10 07:06:26 2026 +0900
[SPARK-57760][SS][PYTHON][FOLLOWUP] Avoid importing numpy at module level
in stateful_processor_api_client
### What changes were proposed in this pull request?
This PR makes `numpy` a lazy import in `stateful_processor_api_client.py`
again. The fix introduces a dedicated `_load_numpy()` function that resolves
numpy availability on first use (called from `_serialize_to_bytes`), and caches
the result in module-level `has_numpy` / `np` variables. The
`_normalize_state_value` function itself remains free of any import or
initialization logic, keeping the hot path lean.
### Why are the changes needed?
SPARK-57760 (#56786) moved `import numpy` to the top level of
`stateful_processor_api_client.py` for micro-optimization. However, this module
is loaded transitively by `import pyspark` via:
import pyspark
-> pyspark.sql
-> pyspark.sql.streaming
-> pyspark.sql.streaming.stateful_processor
-> pyspark.sql.streaming.stateful_processor_api_client
-> import numpy
This causes `test_import_spark_libraries` to fail because the test asserts
that `import pyspark` does not pull in any third-party packages other than py4j.
https://github.com/apache/spark/actions/runs/29027743228/job/86156618984
```
pyspark.tests.test_import_spark with python3.12 failed:
Running tests...
----------------------------------------------------------------------
test_import_spark_libraries
(pyspark.tests.test_import_spark.ImportSparkTest.test_import_spark_libraries)
We want to ensure "import pyspark" is fast. It matters when we spawn ...
FAIL (0.252s)
======================================================================
FAIL [0.252s]: test_import_spark_libraries
(pyspark.tests.test_import_spark.ImportSparkTest.test_import_spark_libraries)
We want to ensure "import pyspark" is fast. It matters when we spawn
----------------------------------------------------------------------
Traceback (most recent call last):
File "/__w/spark/spark/python/pyspark/tests/test_import_spark.py", line
61, in test_import_spark_libraries
self.fail(
AssertionError: Unexpected 3rd party package 'numpy' imported during
'import pyspark'
```
### Does this PR introduce _any_ user-facing change?
No.
### How was this patch tested?
-
`pyspark.tests.test_import_spark.ImportSparkTest.test_import_spark_libraries`
now passes.
- Verified that `_normalize_state_value` correctly normalizes numpy scalars
and nested containers (list/tuple/dict/namedtuple/Row) after `_load_numpy()` is
triggered.
- Confirmed `import pyspark` no longer imports numpy via `-X importtime`.
### Was this patch authored or co-authored using generative AI tooling?
Kiro CLI / Claude
Closes #57163 from sarutak/fix-numpy-import-time.
Authored-by: Kousuke Saruta <[email protected]>
Signed-off-by: Hyukjin Kwon <[email protected]>
---
.../sql/streaming/stateful_processor_api_client.py | 76 +++++++++++++---------
1 file changed, 46 insertions(+), 30 deletions(-)
diff --git a/python/pyspark/sql/streaming/stateful_processor_api_client.py
b/python/pyspark/sql/streaming/stateful_processor_api_client.py
index 166ac52fa6d8..a679c499dc08 100644
--- a/python/pyspark/sql/streaming/stateful_processor_api_client.py
+++ b/python/pyspark/sql/streaming/stateful_processor_api_client.py
@@ -34,36 +34,51 @@ import uuid
__all__ = ["StatefulProcessorApiClient", "StatefulProcessorHandleState"]
-try:
- import numpy as np
-
- has_numpy = True
- SCALAR_TYPES = (bool, int, float, str, bytes, datetime, type(None))
-
- def _normalize_state_value(v: Any) -> Any:
- if type(v) in SCALAR_TYPES: # Fast path for common scalar values.
- return v
- # Convert NumPy scalar values to Python primitive values.
- if isinstance(v, np.generic):
- return v.tolist()
- # Named tuples (collections.namedtuple or typing.NamedTuple) and Row
both
- # require positional arguments and cannot be instantiated with a
generator expression.
- if isinstance(v, Row) or (isinstance(v, tuple) and hasattr(v,
"_fields")):
- return type(v)(*map(_normalize_state_value, v))
- # List / tuple: recursively normalize each element.
- if isinstance(v, (list, tuple)):
- return type(v)(map(_normalize_state_value, v))
- # Dict: normalize both keys and values.
- if isinstance(v, dict):
- return {_normalize_state_value(k): _normalize_state_value(val) for
k, val in v.items()}
- # Address a couple of pandas dtypes too.
- if hasattr(v, "to_pytimedelta"):
- return v.to_pytimedelta()
- if hasattr(v, "to_pydatetime"):
- return v.to_pydatetime()
+# None means not yet checked; True/False after _load_numpy() is called.
+has_numpy: Optional[bool] = None
+np = None
+
+SCALAR_TYPES = (bool, int, float, str, bytes, datetime, type(None))
+
+
+def _load_numpy() -> None:
+ """Lazily resolve numpy availability without importing it at module load
time.
+
+ Importing numpy at the top level would slow down ``import pyspark``
+ (see test_import_spark_libraries).
+ """
+ global has_numpy, np
+ try:
+ import numpy
+
+ np = numpy
+ has_numpy = True
+ except ImportError:
+ has_numpy = False
+
+
+def _normalize_state_value(v: Any) -> Any:
+ if type(v) in SCALAR_TYPES: # Fast path for common scalar values.
return v
-except ImportError:
- has_numpy = False
+ # Convert NumPy scalar values to Python primitive values.
+ if isinstance(v, np.generic):
+ return v.tolist()
+ # Named tuples (collections.namedtuple or typing.NamedTuple) and Row both
+ # require positional arguments and cannot be instantiated with a generator
expression.
+ if isinstance(v, Row) or (isinstance(v, tuple) and hasattr(v, "_fields")):
+ return type(v)(*map(_normalize_state_value, v))
+ # List / tuple: recursively normalize each element.
+ if isinstance(v, (list, tuple)):
+ return type(v)(map(_normalize_state_value, v))
+ # Dict: normalize both keys and values.
+ if isinstance(v, dict):
+ return {_normalize_state_value(k): _normalize_state_value(val) for k,
val in v.items()}
+ # Address a couple of pandas dtypes too.
+ if hasattr(v, "to_pytimedelta"):
+ return v.to_pytimedelta()
+ if hasattr(v, "to_pydatetime"):
+ return v.to_pydatetime()
+ return v
class StatefulProcessorHandleState(Enum):
@@ -520,11 +535,12 @@ class StatefulProcessorApiClient:
return self.utf8_deserializer.loads(self.sockfile)
def _serialize_to_bytes(self, schema: StructType, data: Tuple) -> bytes:
+ if has_numpy is None:
+ _load_numpy()
if has_numpy:
converted = tuple(map(_normalize_state_value, data))
else:
converted = data
-
return self.pickleSer.dumps(schema.toInternal(converted))
def _deserialize_from_bytes(self, value: bytes) -> Any:
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
