[SPARK-52853][SDP] Prevent imperative PySpark methods in declarative pipelines #51590
Conversation
|
addressing discussion |
JiaqiWang18
changed the title
|
@sryza @anishm-db Ready for review. Re-scoped PR to only block imperative python methods on the client-side via a context manager |
python/pyspark/pipelines/tests/test_block_imperative_construct.py
Outdated
Show resolved
Hide resolved
|
|
||
|
|
||
| @contextmanager | ||
| def block_imperative_construct() -> Generator[None, None, None]: |
There was a problem hiding this comment.
| def block_imperative_construct() -> Generator[None, None, None]: | |
| def block_imperative_constructs() -> Generator[None, None, None]: |
There was a problem hiding this comment.
renamed to block_session_mutations to be more clear
| { | ||
| "class": RuntimeConf, | ||
| "method": "set", | ||
| "suggestion": "Instead set configuration via the pipeline spec " |
There was a problem hiding this comment.
It would be better to have this text inside of the error-conditions.json – that way it's in a central place that can be internationalized more easily. Thoughts on having a sub-error code for each of these? E.g. SET_CURRENT_CATALOG?
There was a problem hiding this comment.
Make sense, added sub classes for each method in error-conditons.json
| "Session mutation <method> is not allowed in declarative pipelines." | ||
| ], | ||
| "sub_class": { | ||
| "RUNTIME_CONF_SET": { |
There was a problem hiding this comment.
Nitpick: should the SET be on the other side of RUNTIME_CONF for consistency?
What changes were proposed in this pull request?
This PR adds a context manager
block_imperative_construct()that prevents the execution of imperative Spark operations within declarative pipeline definitions. When these blocked methods are called, users receive clear error messages with guidance on declarative alternatives.Blocked Methods
Configuration Management
spark.conf.set()→ Use pipeline spec orspark_confdecorator parameterCatalog Management
spark.catalog.setCurrentCatalog()→ Set via pipeline spec or dataset decoratornameargumentspark.catalog.setCurrentDatabase()→ Set via pipeline spec or dataset decoratornameargumentTemporary View Management
spark.catalog.dropTempView()→ Remove temporary view definition directlyspark.catalog.dropGlobalTempView()→ Remove temporary view definition directlyDataFrame.createTempView()→ Use@temporary_viewdecoratorDataFrame.createOrReplaceTempView()→ Use@temporary_viewdecoratorDataFrame.createGlobalTempView()→ Use@temporary_viewdecoratorDataFrame.createOrReplaceGlobalTempView()→ Use@temporary_viewdecoratorUDF Registration
spark.udf.register()→ Define and register UDFs before pipeline executionspark.udf.registerJavaFunction()→ Define and register Java UDFs before pipeline executionspark.udf.registerJavaUDAF()→ Define and register Java UDAFs before pipeline executionWhy are the changes needed?
These are imperative construct that can cause friction and unexpected behavior from within a pipeline declaration. E.g. it makes pipeline behavior sensitive to the order that Python files are imported in, which can be unpredictable. There are already existing mechanisms for setting Spark confs for pipelines:
Does this PR introduce any user-facing change?
Yes, it prevents the behavior of setting spark confs imperatively in the pipeline definition file.
How was this patch tested?
Created new test suite to test that the context manager behave as expected and ran
spark-pipelinescli manually.Was this patch authored or co-authored using generative AI tooling?
No