py spark writing invalid date successfully but throwing exception when reading it

user_45280 0
Viewed 241

I'm using py-spark to process some data and writing processed files to S3 with parquet format. Code for the batch process is running on Ec2 in a docker container(Linux). This data contains some date-time fields as well which I save as TimestampType (in parquet file) since I need to support this field in athena queries. If the value of this field is '0001-01-01', batch process successfully writes it to parquet file but only when it reads this data, it throws exception . This is the behavior on linux machine . here is the sample code to repro this - 

from pyspark.sql.types import StructType,StructField,DateType,TimestampType
from dateutil.parser import parse
d=parse('0001-01-01 00:00:00')
data=[{'createdon':d}]
distdata = sc.parallelize(data)
schema = StructType([StructField('createdon',TimestampType())])
df=spark.createDataFrame(distdata,schema)
df.write.parquet("\test-1")

Once this code is executes, it writes the data to file without any error. when I try to read the same, I get below error -

Traceback (most recent call last):                                                                    
  File "<stdin>", line 1, in <module>                                                                 
  File "/usr/local/lib/python3.6/site-packages/pyspark/sql/dataframe.py", line 572, in take           
    return self.limit(num).collect()                                                                  
  File "/usr/local/lib/python3.6/site-packages/pyspark/sql/dataframe.py", line 535, in collect        
    return list(_load_from_socket(sock_info, BatchedSerializer(PickleSerializer())))                  
  File "/usr/local/lib/python3.6/site-packages/pyspark/serializers.py", line 147, in load_stream      
    yield self._read_with_length(stream)                                                              
  File "/usr/local/lib/python3.6/site-packages/pyspark/serializers.py", line 172, in _read_with_length
    return self.loads(obj)                                                                            
  File "/usr/local/lib/python3.6/site-packages/pyspark/serializers.py", line 580, in loads            
    return pickle.loads(obj, encoding=encoding)                                                       
  File "/usr/local/lib/python3.6/site-packages/pyspark/sql/types.py", line 1396, in <lambda>          
    return lambda *a: dataType.fromInternal(a)                                                        
  File "/usr/local/lib/python3.6/site-packages/pyspark/sql/types.py", line 633, in fromInternal       
    for f, v, c in zip(self.fields, obj, self._needConversion)]                                       
  File "/usr/local/lib/python3.6/site-packages/pyspark/sql/types.py", line 633, in <listcomp>         
    for f, v, c in zip(self.fields, obj, self._needConversion)]                                       
  File "/usr/local/lib/python3.6/site-packages/pyspark/sql/types.py", line 445, in fromInternal       
    return self.dataType.fromInternal(obj)                                                            
  File "/usr/local/lib/python3.6/site-packages/pyspark/sql/types.py", line 199, in fromInternal       
    return datetime.datetime.fromtimestamp(ts // 1000000).replace(microsecond=ts % 1000000)           
ValueError: year 0 is out of range

Ideally it should never be written since createdon (datetime) field has invalid value but that is not the behavior. Am I doing something wrong here ?

pyspark
1 Answers

I can not reproduce the issue with the given timestamp, however there are a couple of ways you can validate the timestamp string before writing it:

from pyspark.sql import functions as F
from pyspark.sql import types as T
from dateutil.parser import parse
import datetime


def isvaliddatetime(date_text):
    try:
        datetime.datetime.strptime(date_text, '%Y-%m-%d %H:%M:%S')
        return True
    except ValueError:
        return False


raw_timestamp = "0001-01-01 00:00:00"

assert isvaliddatetime(raw_timestamp)

data=[{'createdon': parse(raw_timestamp)}]
distdata = sc.parallelize(data)
schema = T.StructType([T.StructField('createdon', T.TimestampType())])
df = spark.createDataFrame(distdata, schema)
df.show()

df = df.withColumn("badTimestamp", F.when(F.to_date(F.col("createdon"), "yyyy-MM-dd HH:mm:ss").isNotNull(), False).otherwise(True))
assert df.where(F.col("badTimestamp")==True).count() == 0, f"Corrupt timestamps founds: {[r['createdon'] for r in df.where(F.col('badTimestamp')==True).collect()]}"
df = df.drop("badTimestamp")

df.write.mode('overwrite').parquet(hdfs_root + "\test-so-1")


df_read = spark.read.parquet(hdfs_root + "\test-so-1")
df_read.show()

Maybe the problem is due to datetime, dateutil, or pyspark (I use 2.4.0) version. Interestingly, the above code outputs the following values before and after writing and reading from parquet:

createdon
0001-01-03 00:06:32

which is not the same as the original string, so the value is coerced to a different value. Seems like a bug to me.

user_49587170
Related