The spark driver has stopped unexpectedly and is restarting

Viewed 452

I'm using pyspark on databricks and trying to iterate over a small dataframe (50 rows) using a while loop until the dataframe count is 0 and do some basic manipulations inside the loop (filter, sort, get first row and update some columns) and I get the following error after 15 iterations or so

The spark driver has stopped unexpectedly and is restarting.

After research I found out it's a memory problem.

I'm not using toPandas() or collect(), I'm not using many objects (only 3 dataframes inside the loop and I update them in each iteration), I run the notebook while nothing else is running on the cluster, I tried to increase the driver's memory but I still get the same error. I also tried using gc.collect() and spark.catalog.clearCache() but nothing worked.

The code is something like this:

# df1 and df2 read from the database

columns = ['col1', 'col2', 'col3']

df3 = df1.filter((df1.col2 > 0) & (df1.col3 >= 0))

while df3.count() > 0:

  data = []
  new_data = ['', 0, '']

  df3 = df3.orderBy(df3.col1.asc(), df3.col2.asc())

  df_first = df3.first()

  a = df_first['col1']
  b = df_first['col2']
  c = df_first['col3']

  new_data[0] = a
  new_data[1] = b
  new_data[2] = c

  data.append(tuple(new_data))

  new_df = spark.createDataFrame(data, columns)
  df2 = df2.union(new_df)

  df1 = (df1.withColumn('col2', 
                              when(df1['col1'] == a, df1['col2'] + b)
                              .otherwise(df1['col2'])
                       )
        )

  df3 = (df3.withColumn('col2', 
                              when(df3['col1'] == a, df3['col2'] + b)
                              .otherwise(df3['col2'])
                       )
        )

  df1 = df1.withColumn('col3', df1['col2'] - df1['col4'] - df1['col5'])
  df3 = df3.withColumn('col3', df3['col2'] - df3['col4'] - df3['col5'])

  df3 = df3.filter((df3.col2 > 0) & (df3.col3 >= 0))

Any ideas on what could be causing this issue and how to solve it?

1 Answers

I was able to solve this problem by converting my data frames from pyspark to pandas before the loop and then converting them again to pyspark after the loop.

Turns out you can't use pyspark inside a loop and reassign its value every time, you will get memory issues.

Related