Pfthb

My problem is this.
I have a custom dataframe created from netflow records that contains columns minuteMs (this is the same value for all flows in the given minute), sourcePort and some other values such as sourceIP, packetDelta etc.
There is a row for every flow and there can be many flows for one minute.
What I want to do is compute the number of distinct ports for all flows in the one minute interval, however I just want to count the number of distinct ports that fall within the 90th percentile with regards to how common they are.
This means I just want the number of 90% most distinct ports and the 10% of the least common do not interest me.

What I imagine I need to do is:

1. 
   

   Create a separate daframe for each minute.

   
   
   

   In these resulting dataframes:

   
   


2. Group by port number.


3. Count the number of elements in these groups.


4. Sort these groups by the number of elements.


5. 
   

   Take top 90% of these groups and count how many there are. This will give one value X that is 90th percentile of distinct ports in one minute.

   
   
   

   Back "outside":

   
   


6. Create a one column dataframe that will have as many columns as there are minutes filled with the X values.



7. Join this column with the other dataframe I created by grouping and ordering by Minute and aggregating based on my logic used for computing other values that are not relevant for this question.




8. The final results is a dataframe with a single row for every minute with one of its columns being the desired number of top 90% most common distinct source ports.

The issue is that I am very new to Scala and I have problems even putting together a code for the outlined steps, but more importantly I feel like this is extremely inefficient and can be done much more elegantly, I just dont know how. Any help or pointers is appreciated.

Mock up of an input dataframe:

flowID minuteMs srcPortNumber packetDelta
0 146000 80 5
1 146000 90 8
2 146000 90 4
3 146000 90 4
4 146000 90 17
5 146000 90 22
6 146000 80 2
7 146000 80 3
8 146000 80 4
9 146000 343 3
10 152000 133 5
11 152000 80 31
12 152000 8081 4

Expected output:

minuteMs 90thPercCountDistPorts packetCount
146000 2 72
152000 3 40

The packetCount column is just summed packets from all the flows in the given minute, I dont need help with this it is just there to illustrate that there will be other columns present as well.

Note that port 343 in time 146000 is not counted since it appears only in one flow out of six for that time window, therefore falls into the the least common 10% and as such should not be counted.