Github user rxin commented on a diff in the pull request:
https://github.com/apache/spark/pull/8833#discussion_r39917194
--- Diff:
sql/core/src/main/scala/org/apache/spark/sql/execution/pythonUDFs.scala ---
@@ -338,7 +338,11 @@ case class BatchPythonEvaluation(udf: PythonUDF,
output: Seq[Attribute], child:
def children: Seq[SparkPlan] = child :: Nil
protected override def doExecute(): RDD[InternalRow] = {
- val childResults = child.execute().map(_.copy())
+ val buffer = new
java.util.concurrent.ConcurrentLinkedQueue[InternalRow]()
--- End diff --
If i understand this correctly, we are assuming the following in order for
this to work:
1. Each task gets their own copy of the deserialized closure, and thus
their own copy of the queue.
2. All closures are serialized together in one shot, rather than in
multiple places (e.g. they are all done in the serializer, not in the ctor of
the RDD)
3. Java serializer does not serialize objects twice within the same stream,
since it uses it to detect cycles. When they are deserialized, they still point
to the same copy.
---
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 infrastruct...@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscr...@spark.apache.org
For additional commands, e-mail: reviews-h...@spark.apache.org
