What is an efficient way to apply a relative count based eligibility criteria to Spark dataframe in Scala?

I am pretty new to Scala and I'd appreciate any expert advice. I need to validate several eligibility criteria in specific order on a very large spark data frame (dozens or hundreds of million records). My current approach is very slow and often fails over long running time.

Here is a complete description (with tweak and changes to be sharable): I have a dataset that has several columns (let's call them col1 to 3). Another method looks into these columns and decided what group/subgroups this record is eligible for and they get store in eligibilities column:

+------------------------------------------------------+
| col1 | col2 | col3 | eligibilities                   |
+ -----------------------------------------------------+
|  x1  |  y1  |  z1  | A, a1, a2, B                    |
|  x2  |  y2  |  z2  | A,a1                            |
|  x3  |  y3  |  z3  | A,a2, a3, B, b1                 |
|  x4  |  y4  |  z3  | A,a2                            |
|  x1  |  y2  |  z2  | A,a3,a2,B                       |
|  x2  |  y4  |  z1  | A,a2,B                          |
| ...  | ...  | ...  | ...                             |
+ -----------------------------------------------------+

now given groups A and B, and subgroups a1,a2,a3 and b1. I need to validate in order if each of the subgroups reached the minimum count limit given each row could only be assigned to one subgroup and the count is excluding the one's that have been assigned already. This part of the logic as it requires several counting commands is very slow. To make it easier, I first group by "eligibilities" and count the rows then apply the above logic on the grouped dataset.

+-------------------------------+
| eligibilities       |  count  |
+-------------------------------+
|  A, a1, a2, B       | 120000  |
|  A, a1.             |  43000  |
|  A, a2, a3, B, b1   | 321000  |
|  A,a3,a2,B.         |   5800  |
+-------------------------------+    

Now I apply the decision making logic as follow:

1. filter the groupped dataset by eligibilities to rows that include a1.
2. sum the count column
3. If bigger than threshold. Assign them to a1 and remove the rows else ignore
4. goto 1 and perform for a2,a3,b1,b2 etc.

Here is the general flow of my code:

val subgroups = list("a1","a2","a3","b1")
var dfG = dfx.groupBy("eligibilities").count()

subgroups.foreach { subgroup =>

  val filtered_df = dfG.filter(col("eligibilities").contains(subgroup))
  if (filtered_df(2).count() > 1) {

      val rowCount = filtered_df(col("count")).rdd.map(_ (0).asInstanceOf[Long]).reduce(_ + _)

      if (rowCount > threshold) {

          dfG = dfG.filter(not(col("eligibilities").contains(subgroup)))

      }
  }
}

Is the any better Spark/Scala way to achieve this goal?