下面是如何做到这一点。注意，由于原始Dataframe没有定义顺序，所以顺序没有保留。

import pyspark.sql.functions as F
from pyspark.sql.window import Window

df2 = df.withColumn('ID', F.first('ID', True).over(Window.partitionBy('Code')))

df2.show()
+---+----+
| ID|Code|
+---+----+
|  7|  AZ|
|  7|  AZ|
|  3|  EV|
|  3|  EV|
|  8|  FW|
|  5|  CX|
|  5|  CX|
|  5|  CX|
|  9|  BY|
|  2|  GU|
|  1|  DW|
+---+----+

如果要保持原始顺序，可以执行分配索引的另一个步骤：

df2 = df.withColumn('index', F.monotonically_increasing_id()).withColumn('ID', F.first('ID', True).over(Window.partitionBy('Code'))).orderBy('index').drop('index')

df2.show()
+---+----+
| ID|Code|
+---+----+
|  7|  AZ|
|  5|  CX|
|  9|  BY|
|  5|  CX|
|  5|  CX|
|  1|  DW|
|  7|  AZ|
|  3|  EV|
|  8|  FW|
|  3|  EV|
|  2|  GU|
+---+----+