[jira] [Created] (SPARK-48422) Using lambda may cause MemoryError

["ZhouYang (Jira)" <ji...@apache.org>]

ZhouYang created SPARK-48422:
--------------------------------

             Summary: Using lambda may cause MemoryError
                 Key: SPARK-48422
                 URL: https://issues.apache.org/jira/browse/SPARK-48422
             Project: Spark
          Issue Type: Bug
          Components: PySpark
    Affects Versions: 2.2.1
            Reporter: ZhouYang


In worker.py, there is a function called wrap_udf(f, return_type), lambda is used in its body as below:
{code:java}
def wrap_udf(f, return_type):
    if return_type.needConversion():
        toInternal = return_type.toInternal
        return lambda *a: toInternal(f(*a))
    else:
        return lambda *a: f(*a){code}
Is there any possibility that it will cause MemoryError when function f is used to work on large scale data？For the reason that I have indeed encountered this situation as below:
{code:java}
MemoryError at org.apache.spark.api.python.PythonRunner$$anon$1.read(PythonRDD.scala:203) at org.apache.spark.api.python.PythonRunner$$anon$1.<init>(PythonRDD.scala:244) at org.apache.spark.api.python.PythonRunner.compute(PythonRDD.scala:162) at org.apache.spark.sql.execution.python.BatchEvalPythonExec$$anonfun$doExecute$1.apply(BatchEvalPythonExec.scala:144) at org.apache.spark.sql.execution.python.BatchEvalPythonExec$$anonfun$doExecute$1.apply(BatchEvalPythonExec.scala:87) at org.apache.spark.rdd.RDD$$anonfun$mapPartitions$1$$anonfun$apply$23.apply(RDD.scala:797) at org.apache.spark.rdd.RDD$$anonfun$mapPartitions$1$$anonfun$apply$23.apply(RDD.scala:797) at org.apache.spark.rdd.MapPartitionsRDD.compute(MapPartitionsRDD.scala:38) at org.apache.spark.rdd.RDD.computeOrReadCheckpoint(RDD.scala:323) at org.apache.spark.rdd.RDD.iterator(RDD.scala:287) at org.apache.spark.rdd.MapPartitionsRDD.compute(MapPartitionsRDD.scala:38) at org.apache.spark.rdd.RDD.computeOrReadCheckpoint(RDD.scala:323) at org.apache.spark.rdd.RDD.iterator(RDD.scala:287) at org.apache.spark.scheduler.ResultTask.runTask(ResultTask.scala:89) at org.apache.spark.scheduler.Task.run(Task.scala:109) at org.apache.spark.executor.Executor$TaskRunner$$anon$2.run(Executor.scala:355) at java.security.AccessController.doPrivileged(Native Method) at javax.security.auth.Subject.doAs(Subject.java:422) at org.apache.hadoop.security.UserGroupInformation.doAs(UserGroupInformation.java:1721) at org.apache.spark.executor.Executor$TaskRunner.run(Executor.scala:353) at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149) at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624) at java.lang.Thread.run(Thread.java:748) 2024-05-22 16:50:03,173 INFO org.apache.spark.scheduler.TaskSetManager: Starting task 0.1 in stage 5.0 (TID 21, saturndatanode3, executor 2, partition 0, ANY, 5075 bytes) 2024-05-22 16:50:03,174 INFO org.apache.spark.scheduler.TaskSetManager: Lost task 1.0 in stage 5.0 (TID 19) on saturndatanode3, executor 2: org.apache.spark.api.python.PythonException (Traceback (most recent call last): File "/opt/ZDH/parcels/lib/spark/python/lib/pyspark.zip/pyspark/worker.py", line 200, in main process() File "/opt/ZDH/parcels/lib/spark/python/lib/pyspark.zip/pyspark/worker.py", line 195, in process serializer.dump_stream(func(split_index, iterator), outfile) File "/opt/ZDH/parcels/lib/spark/python/lib/pyspark.zip/pyspark/worker.py", line 106, in <lambda> func = lambda _, it: map(mapper, it) File "<string>", line 1, in <lambda> File "/opt/ZDH/parcels/lib/spark/python/lib/pyspark.zip/pyspark/worker.py", line 73, in <lambda> return lambda *a: f(*a){code}
However, I tried another way with partial instead of lambda in the return sentence, the task completed successfully:
{code:java}
def inner_func(*a):
    f(*a)
return partial(inner_func)
# return lambda *a: f(*a){code}



--
This message was sent by Atlassian Jira
(v8.20.10#820010)

---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@spark.apache.org
For additional commands, e-mail: issues-help@spark.apache.org