pythonapache-sparkpysparktimerexit

Stop Spark Session after some time - Pyspark

I am doing an ETL in spark which sometimes takes a lot of time. I want to gracefully shutdown the spark session after a certain time.

I am writing my code in Pyspark.

try:
 df_final.write.partitionBy("col1","col2","col3").mode("append").format("orc").save(output)
exception:
 spark.stop()

I would like to stop spark after sometime in the above code.

Is there a way to gracefully shutdown the spark session after sometime??

Solution

  • I would suggest use the official python Timer to stop the Spark session gracefully:

    import threading

def timer_elapsed():
    print('Timer elapsed')
    if not sc._jsc.sc().isStopped():
      spark.stop()

# wait for 0.5 sec for Spark job to complete
spark_timer = threading.Timer(0.5, timer_elapsed)
spark_timer.start()

try:
  df_final.write.partitionBy("col1","col2","col3").mode("append").format("orc").save(output
  print('Spark job finished successfully.')
except Exception as e:
  spark_timer.cancel() # stop timer, we don't need to wait if error occured
  if not sc._jsc.sc().isStopped():
    spark.stop()

    Note: We stop the session in two cases if time has elapsed or exception was caught. Before requesting to stop the Spark context we check if the context is active with sc._jsc.sc().isStopped which calls the Java API directly.