I have a set of records for some shipments.
The record set having the columns container_no, origin, destination, shipment_dt and volume.
There are multiple records having the same container_no and having some possibility of that the same container has been shipped on different dates but if the shipment_dt falls within 10 days of time span then check the origin of the container_no, if all the records having different origin then drop all the records having same container_no and falling within the 10 days of time span else select the record with the highest volume.
Pleas Note that: We will decide the 10 days of time span on the basis of the first occurrence of the container_no.
Sample Input:

Origin destination condition of selection data:

I have already written a query to get the time range of ten days but don't know how do I compare the origin and destination and get the records with the highest volume.
Sample Input Query to create dataframe:
val Input_DF = spark.sql("""
SELECT '12345' AS container_no , 'India' AS origin ,'China' AS destination , '2020-10-10' AS shipment_dt , 20 as volume UNION
SELECT '12345' AS container_no , 'India' AS origin ,'China' AS destination , '2020-10-10' AS shipment_dt , 30 as volume UNION
SELECT '12345' AS container_no , 'India' AS origin ,'China' AS destination , '2020-10-12' AS shipment_dt , 10 as volume UNION
SELECT '12345' AS container_no , 'India' AS origin ,'China' AS destination , '2020-10-25' AS shipment_dt , 20 as volume UNION
SELECT '12345' AS container_no , 'India' AS origin ,'China' AS destination , '2020-10-26' AS shipment_dt , 10 as volume UNION
SELECT '12346' AS container_no , 'India' AS origin ,'China' AS destination , '2020-10-15' AS shipment_dt , 20 as volume UNION
SELECT '12346' AS container_no , 'India' AS origin ,'China' AS destination , '2020-10-16' AS shipment_dt , 20 as volume UNION
SELECT '12346' AS container_no , 'India' AS origin ,'China' AS destination , '2020-10-17' AS shipment_dt , 50 as volume UNION
SELECT '12347' AS container_no , 'India' AS origin ,'China' AS destination , '2020-10-18' AS shipment_dt , 20 as volume UNION
SELECT '12347' AS container_no , 'Nepal' AS origin ,'China' AS destination , '2020-10-19' AS shipment_dt , 21 as volume""")
Input_DF.createOrReplaceTempView("Input_DF")
Query to create a 10 days time frame on data:
val output_df = spark.sql("""
SELECT
B.* ,
CASE
WHEN from_prev BETWEEN 0 AND 9
THEN 1
ELSE 0
END AS recent ,
floor(from_first / 10 ) AS recent_group
FROM
(
SELECT
A.*,
NVL(DATEDIFF(shipment_dt,FIRST(shipment_dt) over(partition BY container_no
ORDER BY shipment_dt ASC)) ,0) AS from_first,
NVL(DATEDIFF(shipment_dt,lag(shipment_dt,1) over(partition BY container_no
ORDER BY shipment_dt ASC)) ,0) from_prev
FROM
Input_DF A) B
ORDER BY
container_no,
shipment_dt""")
In the sample input screenshot I have added one extra column to explain that the how one row in different from another records having same container_no but date is different. Thanks in advance.
