[Python] Column dtype returning datetime64[ns] being compared to datetime dtype body

import pandas as pd

rate = spark.sql("SELECT C3, CAST(REPLACE(C4, '.','-') AS timestamp) AS `C4`, C5 from " + src_db_name + ".all_corw WHERE _UNITID = 'S01' AND CODAID = 'ALV'")

rate_pd = rate.toPandas()
print(rate_pd.dtypes)
def addVat(rsum, date, vatcd):

rate_final = rate_pd.loc[(rate_pd['C3']==str(vatcd))&(rate_pd['C4'] < date)].sort_values(by=['C4'], ascending=False)

if rate_final.empty:
result = rsum * Decimal(1 + 0/100)
else:
result = rsum * Decimal(1 + int(rate_final['C5'].iloc[0])/100)

return result

def removeVat(rsum, date, vatcd):

rate_final = rate_pd.loc[(rate_pd['C3']==str(vatcd))&(rate_pd['C4'] < date)].sort_values(by=['C4'], ascending=False)

if rate_final.empty:
result = rsum / Decimal(1 + 0/100)
else:
result = rsum / Decimal(1 + int(rate_final['C5'].iloc[0])/100)

return result

spark.udf.register(name = 'addVat_func', f = addVat, returnType=DecimalType())
spark.udf.register(name = 'removeVat_func', f = removeVat, returnType=DecimalType())


having trouble with this code, returning datetime64[ns] as the dtype of C4, further down the code in this select query

select gb._unitid,max(gb.billd) billd,gs.vehiid,sum(removeVat_func(coalesce(gr.rsum,0),coalesce(gb.billd, cast('1900-01-01' as timestamp)),case when gr.vatcd = '' THEN '0' \
else gr.vatcd end)) summa \
from " + src_db_name + ".all_gbil gb inner join " + src_db_name + ".all_gsal gs on gb.gsalid=gs.gsalid and gb._unitid=gs._unitid \
inner join " + src_db_name + ".all_grow gr on gb.gsalid=gr.gsalid and gb.grecno=gr.grecno and gb._unitid=gr._unitid \
where gb.btype in (30,31,33) and (gr.PACKIDIN like '%GARDX%' OR gr.PACKIDIN like '%GARDEX%' OR gr.PACKIDIN like 'JLRGI') and gb._unitid in (select distinct unitid from unittbl) group by gb._unitid,gs.vehiid \
),


when saving, the code compares the dtype of billd(timestamp) and C4(casted from string to timestamp) ultimately running into Py4JJavaError when the table is being saved. how can i ensure C4 is declared as a datetime dtype instead of the datetime64[ns]. PS. this table is being saved in a unity catalog_schema

Task 1 in stage 4001.0 failed 4 times, most recent failure: Lost task 1.3 in stage 4001.0 (TID 13280) (10.139.64.120 executor 33): org.apache.spark.SparkRuntimeException: [UDF_USER_CODE_ERROR.GENERIC] Execution of function removeVat_func(coalesce(rsum#380619, 0.000000), coalesce(billd#380248, 1900-01-01 00:00:00), CASE WHEN (vatcd#380651 = ) THEN 0 ELSE vatcd#380651 END) failed. == Error == TypeError: Invalid comparison between dtype=datetime64[ns] and datetime


I've tried to avoid using pandas and instead purely used pyspark but ran into a different error stating the provided logic can only run through the driver. The expected should be the dtype of C4 returns with datetime so that when we save the table, C4 and billd pass the dtype comparison.

Continue reading...