Using broadcast values inside PySpark UDF throwing "java.io.IOException: Failed to delete original file"

Viewed 187

Im trying to run my script in local machine and keep getting exception.

Below is issue reproducible smaller version of full script.

Setup: A master and 4 workers(followed this ).

import os
import sys
import math as math
import pickle
import time
import datetime

import pandas as pd
import json
import numpy as np
from pyspark.sql.functions import broadcast
import scipy.linalg as linalg
from scipy import interpolate

from pyspark.context import SparkContext
from pyspark.sql import SparkSession
from pyspark.sql import Window, DataFrame
from pyspark.sql import functions as f
from pyspark.sql.types import IntegralType, StructType, ArrayType, StructField, StringType, DoubleType
import pyspark

@f.pandas_udf(returnType= DoubleType())
def square(r : pd.Series) -> pd.Series:
    offset_value = offset.value
    return (r * r ) + offset_value

if __name__ == "__main__":
    spark = SparkSession.builder.appName("Spark").getOrCreate()

    sc = spark.sparkContext
    
    offset = sc.broadcast(10)
    
    x = pd.Series(np.arange(0,1000000))

    df = spark.createDataFrame(pd.DataFrame(x, columns=["x"]))
    df = df.withColumn('sq',square(df.x)).withColumn('sqsq', square(f.col('sq')))
    start_time = datetime.datetime.now()
    print(f"Show start time {start_time}")
    df.write.mode('overwrite').parquet('output')
    
    end_time = datetime.datetime.now()
    print(f"Show End time {end_time}")
    print(f"Total time taken {end_time - start_time}")
    
    offset.unpersist()

    spark.stop()

Exception:

    Task 1 in stage 0.0 failed 4 times, most recent failure: Lost task 1.3 in stage 0.0 (TID 28, 172.16.1.118, executor 0): java.io.IOException: Failed to delete original file 
'C:\Users\wpf\AppData\Local\Temp\spark-5daa943f-52e5-4d42-9bca-be5a7835da65\executor-f828cf24-85ed-44c1-b0a1-4c310090b657\spark-cbf4b9e6-76f6-4763-904a-3464b89261e8\broadcast8871232241519207229' 
after copy to 'C:\Users\wpf\AppData\Local\Temp\spark-5daa943f-52e5-4d42-9bca-be5a7835da65\executor-f828cf24-85ed-44c1-b0a1-4c310090b657\blockmgr-315d3c00-8817-45ad-b5df-9a866003d8d7\37\broadcast_0_python'

If i remove the offset_value usage inside udf, it is running without issues. but i need broadcasted values inside udf.

Can any one please let me know, what is wrong with my script?

0 Answers
Related