Spark find max of date partitioned column

user_3003363 0
Viewed 5142

I have a parquet partitioned in the following way:

data
/batch_date=2020-01-20
/batch_date=2020-01-21
/batch_date=2020-01-22
/batch_date=2020-01-23
/batch_date=2020-01-24

Here batch_date which is the partition column is of date type.

I want only read the data from the latest date partition but as a consumer I don't know what is the latest value.

I could use a simple group by something like

df.groupby().agg(max(col('batch_date'))).first()

While this would work it's a very inefficient way since it involves a groupby.

I want to know if we can query the latest partition in a more efficient way.

Thanks.

pyspark apache-spark
4 Answers

Doing the method suggested by @pasha701 would involve loading the entire spark data frame with all the batch_date partitions and then finding max of that. I think the author is asking for a way to directly find the max partition date and load only that. One way is to use hdfs or s3fs, and load the contents of the s3 path as a list and then finding the max partition and then loading only that. That would be more efficient.

Assuming you are using AWS s3 format, something like this:

import sys
import s3fs

datelist=[]
inpath="s3:bucket_path/data/"
fs = s3fs.S3FileSystem(anon=False)
Dirs = fs.ls(inpath)
for paths in Dirs:
    date=paths.split('=')[1]
    datelist.append(date)
maxpart=max(datelist)

df=spark.read.parquet("s3://bucket_path/data/batch_date=" + maxpart)

This would do all the work in lists without loading anything into memory until it finds the one you want to load.

user_116185860

Function "max" can be used without "groupBy":

df.select(max("batch_date"))
user_26673430

Using Show partitions to get all partition of table 

show partitions TABLENAME

Output will be like 

pt=2012.07.28.08/is_complete=1
pt=2012.07.28.09/is_complete=1

we can get data form specific partition using below query 

select * from TABLENAME where pt='2012.07.28.10' and is_complete='1' limit 1;

Or additional filter or group by can be applied on it.

user_60858030

This worked for me in Pyspark v2.4.3. First extract partitions (this is for a dataframe with a single partition on a date column, haven't tried it when a table has >1 partitions):

df_partitions = spark.sql("show partitions database.dataframe")

"show partitions" returns dataframe with single column called 'partition' with values like partitioned_col=2022-10-31. Now we create a 'value' column extracting just the date part as string. This is then converted to date and the max is taken:

date_filter = df_partitions.withColumn('value', to_date(split('partition', '=')[1], 'yyyy-MM-dd')).agg({"value":"max"}).first()[0]

date_filter contains the maximum date from the partition and can be used in a where clause pulling from the same table.

user_191893220
Related