How to handle exception in pyspark?

user_13126794 0
Viewed 1583

I have a following code written in pyspark, which basically does certain transformation.

df2=df1.select('*',F.explode(F.col('UseData')).alias('UseData1'))\
           .select('*',F.explode(F.col('UseData1')).alias('UseData2'))\
           .drop('UseData','UseData1','value')\
           .select('*',F.explode(F.col('point'))).drop('point')\
           .withColumn('label',F.col('UseData2.label')).filter(F.col('label')=='jrnyCount')\
           .withColumn('value',F.col('UseData2.value'))\
           .withColumn('datetime',F.col('UseData2.datetime'))\
           .withColumn('latitude',F.col('col.latitude')).withColumn('longitude',F.col('col.longitude'))\
           .drop('col','UseData2')\
           .where(F.col('latitude').isNotNull() | F.col('longitude').isNotNull())

Is there any way to catch if there is any exception occurs due to bad data in input dataframe df1? Since the job is executed in multiple executors across different nodes, how can I make sure if there is any error in any of the above lines, the code will not fail and it ignores the bad data? Any help would be highly apprcitaed.

pyspark apache-spark
1 Answers

Exception handling is to be done with python exception handling methods. But I think what you want is to write some logic to ignore outlier/junk data, that should be done as part of pre-processing manually; may be write a udf to filter or update data based on conditions.

If there are Errors or Warnings when you job is executed, you will get it in the YARN logs so you don't have to handle them specifically at a node level.

user_21429940
Related