[Python] How to validate the latest subdirectories in HDFS using pyspark

I need to check the subdirectory status in HDFS and based on that I need to flag a column. My desired output is:

Date SourceName SourceType IndName IndValue
2024.01.30 PPL app PPL_Repo OK
2024.01.30 INT app INT_Repo OK


The IndValue should have "OK" or "KO" values based on the date folders present inside the path. If the path contains today folder or yesterday folder then its "OK" else "KO" I have multiple dictionary elements to check and update the indicator.

This is my code as of now

dict_kpi = \
{
'PPL_repo': \
{
'path' : '/path/to/ppl/',
'SourceName' : 'PPL',
'SourceType' : 'app',
'format' : 'parquet',
'filter' : None,
'datepattern' : 'date=%Y-%m-%d',
'IndValue' : None,
'IndName' : 'PPL_Repo'
},
'INT_Repo': \
{
'path' : '/path/to/int/',
'SourceName' : 'INT',
'SourceType' : 'app',
'format' : 'json',
'filter' : 'context == sign',
'datepattern' : '%Y-%m-%d',
'IndValue' : None,
'IndName' : 'INT_Repo'
}
}

dfs = []
for _date in [datetime(2024,01,30)]:
for j in dict_kpi.values():
cols_selected = \
[
F.lit(str(_date)[0:10]).alias('Date'),
F.lit(j['SourceName']).alias('SourceName'),
F.lit(j['SourceType']).alias('SourceType'),
F.lit(j['IndName']).alias('IndName'),
j['IndValue'].alias('IndValue')
]
dfs.append(spark.read.format(j['format']).load(os.path.join(j['path'],
_date.strftime(j['datepattern']))).select(*cols_selected))
dfs[0].show()



This is not working as expected. I will have many elements in the dictionary with different path, fileformat, datepattern and different filters

If difference between the two dates is more than 1 day, apply a "KO" flag. Otherwise check the presence of filter. If exists then "OK" else "KO".

What changes are required for this.

Continue reading...