I am unable to send broadcast variable of size more than 1MB to udf. Here is the code
from pyspark.sql.functions import udf, col
import numpy as np
spark = SparkSession.builder.appName("Spark").getOrCreate()
sc = spark.sparkContext
d = [('Alice', 10), ('bob', 21), ('Met', 25), ('Andy', 27)] # data frame rows
df = spark.createDataFrame(d , ["name", "age"])
l = [50000] * np.ones((1300, 100))
rdd = sc.parallelize(l)
b_rdd = sc.broadcast(rdd.collect())
def test (x, data):
print(x)
print(len(data))
def udf_test(data):
return udf(lambda x : test(x, data))
df.withColumn("test", udf_test(b_rdd.value)(col("age"))).show()
spark.stop()
when i run this:
Submit-Job.cmd --master spark://xx.xx.0.24:7077 --executor-memory=20G test_2.py
I get the below error:
java.io.IOException: Failed to delete original file 'C:\Users\h230236\AppData\Local\Temp\spark-dceed17c-fc67-43cb-9e99-69571cd36020\executor-33071657-26d7-4e89-b55a-9a8bc58b471e\spark-a184bb5a-9632-48ad-8364-99d53a0b398a\broadcast1428986508512125101' after copy to 'C:\Users\h230236\AppData\Local\Temp\spark-dceed17c-fc67-43cb-9e99-69571cd36020\executor-33071657-26d7-4e89-b55a-9a8bc58b471e\blockmgr-852e475e-890c-417d-88c3-0ed25b77e4dc\39\broadcast_2_python'
at org.apache.commons.io.FileUtils.moveFile(FileUtils.java:2835)
at org.apache.spark.storage.DiskStore.moveFileToBlock(DiskStore.scala:133)
at org.apache.spark.storage.BlockManager$TempFileBasedBlockStoreUpdater.saveToDiskStore(BlockManager.scala:424)
at org.apache.spark.storage.BlockManager$BlockStoreUpdater.$anonfun$save$1(BlockManager.scala:343)
at org.apache.spark.storage.BlockManager.org$apache$spark$storage$BlockManager$$doPut(BlockManager.scala:1298)
at org.apache.spark.storage.BlockManager$BlockStoreUpdater.save(BlockManager.scala:316)
at org.apache.spark.storage.BlockManager$TempFileBasedBlockStoreUpdater.save(BlockManager.scala:427)
at org.apache.spark.api.python.PythonBroadcast.$anonfun$readObject$2(PythonRDD.scala:771)
at scala.runtime.java8.JFunction0$mcZ$sp.apply(JFunction0$mcZ$sp.java:23)
at org.apache.spark.util.Utils$.tryWithSafeFinally(Utils.scala:1377)
at org.apache.spark.api.python.PythonBroadcast.$anonfun$readObject$1(PythonRDD.scala:773)
at scala.runtime.java8.JFunction0$mcZ$sp.apply(JFunction0$mcZ$sp.java:23)
at org.apache.spark.util.Utils$.tryOrIOException(Utils.scala:1343)
at org.apache.spark.api.python.PythonBroadcast.readObject(PythonRDD.scala:760)
at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
at sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62)
at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
at java.lang.reflect.Method.invoke(Method.java:498)
at java.io.ObjectStreamClass.invokeReadObject(ObjectStreamClass.java:1158)
at java.io.ObjectInputStream.readSerialData(ObjectInputStream.java:2169)
at java.io.ObjectInputStream.readOrdinaryObject(ObjectInputStream.java:2060)
at java.io.ObjectInputStream.readObject0(ObjectInputStream.java:1567)
at java.io.ObjectInputStream.readObject(ObjectInputStream.java:427)
at org.apache.spark.serializer.JavaDeserializationStream.readObject(JavaSerializer.scala:76)
at org.apache.spark.broadcast.TorrentBroadcast$.$anonfun$unBlockifyObject$4(TorrentBroadcast.scala:328)
at org.apache.spark.util.Utils$.tryWithSafeFinally(Utils.scala:1377)
at org.apache.spark.broadcast.TorrentBroadcast$.unBlockifyObject(TorrentBroadcast.scala:330)
at org.apache.spark.broadcast.TorrentBroadcast.$anonfun$readBroadcastBlock$4(TorrentBroadcast.scala:249)
at scala.Option.getOrElse(Option.scala:189)
at org.apache.spark.broadcast.TorrentBroadcast.$anonfun$readBroadcastBlock$2(TorrentBroadcast.scala:223)
at org.apache.spark.util.KeyLock.withLock(KeyLock.scala:64)
at org.apache.spark.broadcast.TorrentBroadcast.$anonfun$readBroadcastBlock$1(TorrentBroadcast.scala:218)
at org.apache.spark.util.Utils$.tryOrIOException(Utils.scala:1343)
at org.apache.spark.broadcast.TorrentBroadcast.readBroadcastBlock(TorrentBroadcast.scala:218)
at org.apache.spark.broadcast.TorrentBroadcast.getValue(TorrentBroadcast.scala:103)
at org.apache.spark.broadcast.Broadcast.value(Broadcast.scala:70)
at org.apache.spark.api.python.BasePythonRunner$WriterThread.$anonfun$run$20(PythonRunner.scala:374)
at scala.collection.immutable.Stream.foreach(Stream.scala:533)
at org.apache.spark.api.python.BasePythonRunner$WriterThread.$anonfun$run$1(PythonRunner.scala:370)
at org.apache.spark.util.Utils$.logUncaughtExceptions(Utils.scala:1932)
at org.apache.spark.api.python.BasePythonRunner$WriterThread.run(PythonRunner.scala:218)
20/11/06 17:01:04 ERROR PythonUDFRunner: Python worker exited unexpectedly (crashed)
org.apache.spark.api.python.PythonException: Traceback (most recent call last):
File "C:\NMR_PySpark\spark-3.0.0-bin-hadoop2.7\python\lib\pyspark.zip\pyspark\worker.py", line 467, in main
File "C:\NMR_PySpark\spark-3.0.0-bin-hadoop2.7\python\lib\pyspark.zip\pyspark\serializers.py", line 595, in read_int
raise EOFError
EOFError
at org.apache.spark.api.python.BasePythonRunner$ReaderIterator.handlePythonException(PythonRunner.scala:503)
at org.apache.spark.sql.execution.python.PythonUDFRunner$$anon$2.read(PythonUDFRunner.scala:81)
at org.apache.spark.sql.execution.python.PythonUDFRunner$$anon$2.read(PythonUDFRunner.scala:64)
at org.apache.spark.api.python.BasePythonRunner$ReaderIterator.hasNext(PythonRunner.scala:456)
at org.apache.spark.InterruptibleIterator.hasNext(InterruptibleIterator.scala:37)
at scala.collection.Iterator$$anon$11.hasNext(Iterator.scala:489)
at scala.collection.Iterator$$anon$10.hasNext(Iterator.scala:458)
at scala.collection.Iterator$$anon$10.hasNext(Iterator.scala:458)
at org.apache.spark.sql.catalyst.expressions.GeneratedClass$GeneratedIteratorForCodegenStage2.processNext(Unknown Source)
at org.apache.spark.sql.execution.BufferedRowIterator.hasNext(BufferedRowIterator.java:43)
at org.apache.spark.sql.execution.WholeStageCodegenExec$$anon$1.hasNext(WholeStageCodegenExec.scala:729)
at org.apache.spark.sql.execution.SparkPlan.$anonfun$getByteArrayRdd$1(SparkPlan.scala:340)
at org.apache.spark.rdd.RDD.$anonfun$mapPartitionsInternal$2(RDD.scala:872)
at org.apache.spark.rdd.RDD.$anonfun$mapPartitionsInternal$2$adapted(RDD.scala:872)
at org.apache.spark.rdd.MapPartitionsRDD.compute(MapPartitionsRDD.scala:52)
at org.apache.spark.rdd.RDD.computeOrReadCheckpoint(RDD.scala:349)
at org.apache.spark.rdd.RDD.iterator(RDD.scala:313)
at org.apache.spark.scheduler.ResultTask.runTask(ResultTask.scala:90)
at org.apache.spark.scheduler.Task.run(Task.scala:127)
at org.apache.spark.executor.Executor$TaskRunner.$anonfun$run$3(Executor.scala:444)
at org.apache.spark.util.Utils$.tryWithSafeFinally(Utils.scala:1377)
at org.apache.spark.executor.Executor$TaskRunner.run(Executor.scala:447)
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)
Caused by: java.io.IOException: Failed to delete original file 'C:\Users\h230236\AppData\Local\Temp\spark-dceed17c-fc67-43cb-9e99-69571cd36020\executor-33071657-26d7-4e89-b55a-9a8bc58b471e\spark-a184bb5a-9632-48ad-8364-99d53a0b398a\broadcast1428986508512125101' after copy to 'C:\Users\h230236\AppData\Local\Temp\spark-dceed17c-fc67-43cb-9e99-69571cd36020\executor-33071657-26d7-4e89-b55a-9a8bc58b471e\blockmgr-852e475e-890c-417d-88c3-0ed25b77e4dc\39\broadcast_2_python'
However if I use this data to broadcast
l = [50000] * np.ones((1200, 100))
which is less than 1MB, and it runs with no error.
+-----+---+----+
| name|age|test|
+-----+---+----+
|Alice| 10|null|
| bob| 21|null|
| Met| 25|null|
| Andy| 27|null|
+-----+---+----+
Can you please help me to increase the broadcast variable size limit I tried with this option, but didn't help
Submit-Job.cmd --master spark://xx.xx.0.24:7077 --executor-memory=20G --conf spark.broadcast.blockSize=100m test_1.py