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