need to find the latest records for a composite keys in spark sql

0
votes

I need to find the latest records of full_national_number based on the date. Can someone please suggest the solution?

My data is

+--------------------+-----------------------+----------+ |full_national_number|derived_sequence_number| ts| +--------------------+-----------------------+----------+ | A00000001 | 0000|1111-11-11| | A00000001 | 0001|1111-11-11| | A00000001 | 0002|1111-11-11| | A00000002 | 0000|1111-11-11| | A00000002 | 0001|1111-11-11| | A00000002 | 0002|1111-11-11| | A00000003 | 0000|1111-11-11| | A00000003 | 0001|1111-11-11| | A00000004 | 0000|1111-11-11| | A000000010 | 0000|1111-11-11| | A000000011 | 0000|1111-11-11| | A00000008 | 0000|2018-11-16| | A00000008 | 0001|2018-11-16| | A00000008 | 0002|2018-11-16| | A00000002 | 0000|2018-11-16| | A00000003 | 0000|2018-11-16| | A00000004 | 0000|2018-11-16| | A00000005 | 0000|2018-11-16| +--------------------+-----------------------+----------+

My expected output should be 

+--------------------+-----------------------+----------+
|full_national_number|derived_sequence_number|        ts|
+--------------------+-----------------------+----------+
|    A00000001       |                   0000|1111-11-11|
|     A00000001      |                   0001|1111-11-11|
|    A00000001       |                   0002|1111-11-11|
|    A00000002       |                   0000|2018-11-16|
|    A00000003       |                   0000|2018-11-16|
|    A00000004       |                   0000|2018-11-16|
|    A00000005       |                   0000|2018-11-16|
|    A00000008       |                   0000|2018-11-16|
|    A00000008       |                   0001|2018-11-16|
|    A00000008       |                   0002|2018-11-16|
|   A000000010       |                   0000|1111-11-11|
|   A000000011       |                   0000|1111-11-11|
+--------------------+-----------------------+----------+

I have tried the below option but got an error.

sqlContext.sql("select full_national_number, derived_sequence_number, max(ts) from (select *,to_date('1111-11-11') as ts from t1 union all select *,current_date from t2) unioned group by full_national_number").show()

The error I get is

Traceback (most recent call last):
File "", line 1, in (module) File "/opt/cloudera/parcels/CDH-5.14.4-1.cdh5.14.4.p0.3/lib/spark/python/pyspark/sql/context.py", line 580, in sql return DataFrame(self._ssql_ctx.sql(sqlQuery), self)
File "/opt/cloudera/parcels/CDH-5.14.4-1.cdh5.14.4.p0.3/lib/spark/python/lib/py4j-0.10.7-src.zip/py4j/java_gateway.py", line 1257, in call
File "/opt/cloudera/parcels/CDH-5.14.4-1.cdh5.14.4.p0.3/lib/spark/python/pyspark/sql/utils.py", line 51, in deco raise AnalysisException(s.split(': ', 1)[1], stackTrace) pyspark.sql.utils.AnalysisException: u"expression 'derived_sequence_number' is neither present in the group by, nor is it an aggregate function. Add to group by or wrap in first() (or first_value) if you don't care which value you get.;"

Please suggest me a solution for this.

1
pythonsqlpysparkpyspark-sql

1 Answers

1
votes

I think this will get you the results you want. Just pasting the SQL query:

Select full_national_number, derived_sequence_number, ts
FROM
(
select full_national_number, derived_sequence_number, ts, 
RANK() OVER(Partition by full_national_number ORDER by ts desc) as rnk
from table
)a
WHERE a.rnk = 1;

Let me know if this helps.