Создание столбца, который считает количество повторений в кадрах данных spark

Question

У меня есть этот большой фрейм данных, длиной 7 миллионов строк, и мне нужно добавить столбец, который подсчитывает, сколько раз определенное лицо (идентифицированное и Integer) появлялось раньше, например:

| Reg |   randomdata |                   
| 123 | yadayadayada |                 
| 246 | yedayedayeda |          
| 123 | yadeyadeyade |                 
|369  | adayeadayead |                
| 123 | yadyadyadyad |  

до ->

| Reg |   randomdata |     count              
| 123 | yadayadayada |          1       
| 246 | yedayedayeda |          1  
| 123 | yadeyadeyade |          2      
| 369 | adayeadayead |          1      
| 123 | yadyadyadyad |          3

Я уже сделал groupBy, чтобы узнать, сколько раз каждое повторяется, но мне нужно получить этот счет для упражнения по машинному обучению, чтобы получить вероятность повторения в зависимости от того, сколько раз это было раньше.

Answer 1

Следующее, при котором мы предполагаем, что случайность может означать то же случайное значение, возникающее и использующее spark sql с tempview, но может быть сделано также с DF с select:

import org.apache.spark.sql.functions._
import org.apache.spark.sql.expressions.Window._

case class xyz(k: Int, v: String)
val ds = Seq(
     xyz(1,"917799423934"),
     xyz(2,"019331224595"),
     xyz(3,"8981251522"),
     xyz(3,"8981251522"),
     xyz(4,"8981251522"),
     xyz(1,"8981251522"),
     xyz(1,"uuu4553")).toDS()

 ds.createOrReplaceTempView("XYZ")

spark.sql("""select z.k, z.v, dense_rank() over (partition by z.k order by z.seq) as seq from (select k,v, row_number() over (order by k) as seq from XYZ) z""").show

возвращение:

+---+------------+---+
|  k|           v|seq|
+---+------------+---+
|  1|917799423934|  1|
|  1|  8981251522|  2|
|  1|     uuu4553|  3|
|  2|019331224595|  1|
|  3|  8981251522|  1|
|  3|  8981251522|  2|
|  4|  8981251522|  1|
+---+------------+---+

Answer 2

Вы можете сделать что-то вроде этого

 def countrds = udf((rds: Seq[String]) => {rds.length})
 val df2 = df1.groupBy(col("Reg")).agg(collect_list(col("randomdata")).alias("rds"))
                    .withColumn("count", countrds(col("rds")))
 df2.select('Reg', 'randomdata', 'count').show()