BUG: Add temporary struct col in pyspark backend to ensure that UDFs are e… #2657
Conversation
icexelloss
force-pushed
the
pyspark-udf-struct-selection
branch
from
6d0c322
to
30b05c3
Compare
jreback
added
expressions
icexelloss
added
pyspark
icexelloss
changed the title
icexelloss
changed the title
|
@jreback Added tests. The change is for spark only. And new tests are for both pandas and spark. |
| output_type=dt.Struct(['col1', 'col2'], [dt.double, dt.double]), | ||
| ) | ||
| def add_one_struct_exact_once(v): | ||
| print(v) |
There was a problem hiding this comment.
Aha my bad. Removed.
| print(v) | ||
| key = v.iloc[0] | ||
| path = Path(f"{tempdir}/{key}") | ||
| assert not path.exists() |
There was a problem hiding this comment.
why are you writing things?
There was a problem hiding this comment.
This is basically to create a side effect. If this function gets run for the second time it will hit the side effect and fail.
There was a problem hiding this comment.
ok cool (may want to document this in the future)
|
|
||
| import pyspark | ||
|
|
||
| if LooseVersion(pyspark.__version__) < LooseVersion("3.1.1"): |
There was a problem hiding this comment.
hmm not sure i love this here
can you create a pytest decorator and use it here, similar to this: https://github.com/pandas-dev/pandas/blob/master/pandas/util/_test_decorators.py#L200 (you can use skipif, but have to do the import too)
There was a problem hiding this comment.
Added marker min_spark_version
| print(v) | ||
| key = v.iloc[0] | ||
| path = Path(f"{tempdir}/{key}") | ||
| assert not path.exists() |
There was a problem hiding this comment.
ok cool (may want to document this in the future)
|
thanks @icexelloss |
…xecuted once
What change is proposed
Spark sometimes execute the UDF multiple times if
This change ensures that struct column is always assigned to the Spark DataFrame before destructing to ensure exact once execution.
How is this tested
Add test for exact once execution verification under
test_vectorized_udf.py