Добавить перезаписывающий индекс в спрей данных

Question

У меня есть фрейм данных. Я хотел бы добавить индексный столбец для сброса на основе одного из столбцов

--------------------
|  ColA   |  ColB  |
====================
|  G1     |  10    |
--------------------
|  G1     |  20    |
--------------------
|  G2     |  50    |
--------------------
|  G2     |  10    |
--------------------
|  G2     |  70    |
--------------------

Я хотел бы, чтобы результат был

-----------------------------
|  ColA   |  ColB  |  ColC  |
=============================
|  G1     |  10    |   1    |
-----------------------------
|  G1     |  20    |   2    |
-----------------------------
|  G2     |  50    |   1    |   <== reset because ColA changed
-----------------------------
|  G2     |  10    |   2    |
-----------------------------
|  G2     |  70    |   3    |
-----------------------------

Есть ли что-нибудь вроде df.withColumn ("id", monotonicallyIncreasingId), которое будет уместным?

Answer 1

Используйте Window для создания разделов для столбца colA.

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

val w = Window.partitionBy("ColA").orderBy("ColB")
df.withCloumn("id", row_number.over(w))

Или, если вы хотите сохранить исходный порядок строк,

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

val w = Window.partitionBy("ColA").orderBy("temp")
df.withColumn("temp", monotonically_increasing_id)
  .withCloumn("id", row_number.over(w))
  .drop("temp")