Github user JoshRosen commented on the pull request:
https://github.com/apache/spark/pull/2838#issuecomment-59632902
The `take()`-handling code is a little tricky and has been the source of
multiple bugs in the past, so it may be worth considering several different
approaches for fixing this; I would hate for this problem to re-occur again.
For those who are unfamiliar with this code / issue, here a quick summary
of this issue:
- When I call `take()` in PySpark, this translates down to a
`mapPartitions()` call where the map function returns the `take()` results
before consuming all of the partition's iterator.
- As a result, Python workers can complete their tasks before their input
writers are closed.
- Previously, this could lead to various IOExceptions because the Java
thread that writes the Python worker's input would still be running when the
Python worker process shut down on task completion.
- Now that we re-use workers, the unread data written by the Java data
writer thread will be left in the socket. This causes problems for subsequent
tasks run on that worker, since the worker attempts to interpret whatever
random data was left in the socket as Python code, configuration settings, etc.
- This bug, like the old IOException issue, only surfaces in cases where we
have a large partition and a small `take()` size. If the partition is small or
the `take()` is big, this stdin writer thread's output may be completely read
by the worker. Basically, there's a race condition here which is data-size /
take-size dependent.
This PR proposes to fix this by having the worker check the input stream to
determine whether there was unread data left in the stream. If so, the worker
exits and is not re-used.
I have two questions:
- Is there an approach that allows the worker to be (safely) re-used?
- Is there a less-brittle way to check whether the worker has consumed all
of its input? I think we might be able to do this by calling `next()` on the
input stream after we have run the user-provided code and checking for the
"end-of-stream" code there, or checking for it inside of the serializers
themselves. As I explained in [my comment in the
diff](https://github.com/apache/spark/pull/2838#discussion_r19056687), I think
that this may be less brittle in the face of future changes to our
serialization code.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at [email protected] or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
