Select Data from multiple rows to one row

user_4998778 0
Viewed 85

I'm pretty new to functional programming and pyspark and I currently struggle to condense the data I want from my source data

Let's say I have two tables as DataFrames:

# if not already created automatically, instantiate Sparkcontext
spark = SparkSession.builder.getOrCreate()

columns = ['Id', 'JoinId', 'Name']
vals = [(1, 11, 'FirstName'), (2, 12, 'SecondName'), (3, 13, 'ThirdName')]
persons = spark.createDataFrame(vals,columns)
columns = ['Id', 'JoinId', 'Specification', 'Date', 'Destination']
vals = [(1, 10, 'I', '20051205', 'New York City'), (2, 11, 'I', '19991112', 'Berlin'), (3, 11, 'O', '20030101', 'Madrid'), (4, 13, 'I', '20200113', 'Paris'), (5, 11, 'U', '20070806', 'Lissabon')]
movements = spark.createDataFrame(vals,columns)

persons.show()
+---+------+----------+
| Id|JoinId|      Name|
+---+------+----------+
|  1|    11| FirstName|
|  2|    12|SecondName|
|  3|    13| ThirdName|
+---+------+----------+

movements.show()
+---+------+-------------+--------+-------------+
| Id|JoinId|Specification|    Date|  Destination|
+---+------+-------------+--------+-------------+
|  1|    10|            I|20051205|New York City|
|  2|    11|            I|19991112|       Berlin|
|  3|    11|            O|20030101|       Madrid|
|  4|    13|            I|20200113|        Paris|
|  5|    11|            U|20070806|     Lissabon|
+---+------+-------------+--------+-------------+

What I want to create is

+--------+----------+---------+---------+-----------+
|PersonId|PersonName|    IDate|    ODate|Destination|
|       1| FirstName| 19991112| 20030101|     Berlin|
|       3| ThirdName| 20200113|         |      Paris|
+--------+----------+---------+---------+-----------+

The rules would be:

  1. PersonId is the Id of the Person
  2. IDate is the Date saved in the Movements DataFrame where Specification is I
  3. ODate the Date saved in the Movements DataFrame where Specification is O
  4. The Destination is the Destination of the joined entry where the Specification was I

I already joined the dataframes on JoinId

joined = persons.withColumnRenamed('JoinId', 'P_JoinId').join(movements, col('P_JoinId') == movements.JoinId, how='inner')

joined.show()
+---+--------+---------+---+------+-------------+--------+-----------+
| Id|P_JoinId|     Name| Id|JoinId|Specification|    Date|Destination|
+---+--------+---------+---+------+-------------+--------+-----------+
|  1|      11|FirstName|  2|    11|            I|19991112|     Berlin|
|  1|      11|FirstName|  3|    11|            O|20030101|     Madrid|
|  1|      11|FirstName|  5|    11|            U|20070806|   Lissabon|
|  3|      13|ThirdName|  4|    13|            I|20200113|      Paris|
+---+--------+---------+---+------+-------------+--------+-----------+

But I'm struggling to select data from multiple rows and put them with the given rules into a single row...

Thank you for your help

pyspark
1 Answers

Note : I have renamed the id in movements to Id_Movements,to avoid confusion in grouping later.

You can pivot your joined data based on the specification and do some aggregation on date and destination. Then you will get the date and destination specification wise.

import pyspark.sql.functions as F
persons =sqlContext.createDataFrame( [(1, 11, 'FirstName'), (2, 12, 'SecondName'), (3, 13, 'ThirdName')],schema=['Id', 'JoinId', 'Name'])
movements=sqlContext.createDataFrame([(1, 10, 'I', '20051205', 'New York City'), (2, 11, 'I', '19991112', 'Berlin'), (3, 11, 'O', '20030101', 'Madrid'), (4, 13, 'I', '20200113', 'Paris'), (5, 11, 'U', '20070806', 'Lissabon')],schema=['Id_movements', 'JoinId', 'Specification', 'Date', 'Destination'])
df_joined = persons.withColumnRenamed('JoinId', 'P_JoinId').join(movements, F.col('P_JoinId') == movements.JoinId, how='inner')
#%%
df_pivot = df_joined.groupby(['Id','Name']).pivot('Specification').agg(F.min('Date').alias("date"),F.min('Destination').alias('destination'))

Here I have chosen the min aggregation, but you can choose the one as per your need and drop the irrelevant columns

results :

+---+---------+--------+-------------+--------+-------------+--------+-------------+
| Id|     Name|  I_date|I_destination|  O_date|O_destination|  U_date|U_destination|
+---+---------+--------+-------------+--------+-------------+--------+-------------+
|  1|FirstName|19991112|       Berlin|20030101|       Madrid|20070806|     Lissabon|
|  3|ThirdName|20200113|        Paris|    null|         null|    null|         null|
+---+---------+--------+-------------+--------+-------------+--------+-------------+
user_112378340
Related