title: Spark SQL Group by and evaluate duplicates rows
body: <p>I'm begginner working with Spark SQL in Java API.
  I have a dataset with duplicate clients grouped by ENTITY and DOCUMENT_I like this:</p>
  <pre><code>.withColumn( &quot;ROWNUMBER&quot;, row_number().over(Window.partitionBy(&quot;ENTITY&quot;, &quot;ENTITY_DOC&quot;).orderBy(&quot;ID&quot;)))
  </code></pre>
  <pre><code>+---------+----------+-----------------+----------+-----------+-----------+--------+------------+--------------------------+
  |ROWNUMBER|  ENTITY  |      ENTITY_DOC |    ID    |  BLOCK    |  TYPE_DOC |COD_BEST|COD_CRITERIO|AUD_TIMESTAM              |
  +---------+----------+-----------------+----------+-----------+-----------+--------+------------+--------------------------+
  |        1|       182|000004693R       |   5254578|          0|         01|       0|           0|2005-07-07 01:03:25.613802|
  |        2|       182|000004693R       |  99841470|          1|         02|       0|           0|2005-07-07 01:03:25.613802|
  |        3|       182|000004693R       |  45866239|          3|         01|       0|           0|2000-01-16 09:07:00       |
  |        1|       182|000081638B       |  99804050|          0|         01|       0|           0|2007-03-08 14:37:11.73772 |
  |        2|       182|000081638B       |  99803968|          0|         06|       0|           0|2007-08-28 19:15:27.761901|
  |        3|       182|000081638B       |  99803958|          2|         05|       0|           0|2001-04-30 16:23:23.546574|
  |        4|       182|000081638B       |  99804054|          0|         01|       0|           0|2006-09-25 11:57:57.829207|
  |        5|       182|000081638B       |  99787706|          1|         03|       0|           0|2000-05-06 21:06:00       |
  |        6|       182|000081638B       |  99803930|          0|         01|       0|           0|2007-02-23 11:20:36.628582|
  |        1|       182|000107084L       |  99819126|          0|         04|       0|           0|2007-02-23 11:20:36.628582|
  |        2|       182|000107084L       |  99818446|          2|         01|       0|           0|2000-01-16 09:08:56.417506|
  +---------+----------+-----------------+----------+-----------+-----------+--------+------------+--------------------------+
  
  </code></pre>
  <p>I added a ROWNUMBER column to know how many clients are there (and I have to compare) for each group. The idea is to evaluate the criteria between the first row and the next, until the group is finished and the best is obtained.</p>
  <p>It is decided which is the best based on certain business criteria like:</p>
  <pre><code>//criteria 1 (the best is the ID with BLOCK 0)
  BLOCK = 0 or ' ' VS BLOCK 1, 2 or 3
  
  //criteria 2 (the best is the ID with BLOCK 2)
  BLOCK = 2 VS BLOCK 1 or 3
  
  //criteria3 (the best is the ID with BLOCK 3)
  BLOCK = 3 VS BLOCK 1
  
  
  //criteria4 (the best is the one with the most recent timestamp)
  BLOCKS 1-1  or  BLOCKS 3-3
  
  ** if BLOCKS 2-2 OR BLOCKS 0-0 evaluate the following criteria
  
  //criteria5 (the best is ID with typedoc 01, 02, 03, 04 or 07
  TYPE DOC 01, 02, 03, 04, 07 VS ANY OTHER
  
  //criteria6 (the best is the one with the most recent timestamp)
  TYPE DOC 01, 02, 03, 04, 07 VS TYPE DOC 01, 02, 03, 04, 07
  
  </code></pre>
  <p>Now I have to compare in each group, pairs of rows in order to decide which is the best. <strong>This is the big problem, I dont know if its better create an array struct, a map...</strong>
  I dont know how iterate each row inside the same group. An example:</p>
  <pre><code>if entity &quot;182&quot; with entity_doc &quot;000004693R&quot; have 3 matches, we have to compare:
  
  First, ROW(1) -&gt; ID 5254578 against  ROW(2) ID -&gt; 99841470.
  ** ROW(1) would be the best because criteria 1. 
  
  Following the order, we have to compare the before best ROW(1) -&gt; ID 5254578 VS ROW(3) ID -&gt; 45866239.
  ** ROW(1) would be the best because criteria 1.
   
  </code></pre>
  <p>I tried to group each row of the group in a collect_list but I don't know how to access each element to compare it with the next:</p>
  <pre><code>root
   |-- ENTITY: string (nullable = true)
   |-- ENTITY_DOC: string (nullable = true)
   |-- COMBINED_LIST: array (nullable = true)
   |    |-- element: struct (containsNull = false)
   |    |    |-- ID: string (nullable = true)
   |    |    |-- BLOCK: string (nullable = true)
   |    |    |-- TYPE_DOC: string (nullable = true)
   |    |    |-- AUD_TIMESTAM: timestamp (nullable = true)
  
  Example of first group:
  +------+-------------------------+--------------------------------------------------------------------------------------------------------------------------------------+
  |ENTITY|ENTITY_DOC               |COMBINED_LIST                                                                                                                         |
  +------+-------------------------+--------------------------------------------------------------------------------------------------------------------------------------+
  |182   |000004693R               |[[5254578,0,01, 2005-07-07 01:03:25.613802], [99841470, 1,02, 2005-07-07 01:03:25.613802], [45866239, 3,01, 2000-01-16 09:07:00]]     |
  </code></pre>
  <p>The output would be a TUPLE with the best ID and the others ID's LIKE:</p>
  <pre><code>+----------+-----------------+----------+----------+------------+
  |  ENTITY  |      ENTITY_DOC |    ID1   |    ID2   |COD_CRITERIA|
  +----------+-----------------+----------+----------+------------+
  |       182|000004693R       |   5254578|  99841470|           1|
  |       182|000004693R       |   5254578|  45866239|           1|
  
  
  </code></pre>
  <p>Can you help me with Spark?
  Thanks a lot!</p>