如果我正确理解了您的问题，那么您希望在分组数据上复制stringindexer（）的行为。这样做（用英语） pySpark )，我们首先定义 udf 它将在一个 List 包含每个组的所有值的列。请注意，计数相等的元素将被任意排序。

from collections import Counter
from pyspark.sql.types import ArrayType, IntegerType

def encoder(col):

  # Generate count per letter
  x = Counter(col)

  # Create a dictionary, mapping each letter to its rank
  ranking = {pair[0]: rank 
           for rank, pair in enumerate(x.most_common())}

  # Use dictionary to replace letters by rank
  new_list = [ranking[i] for i in col]

  return(new_list)

encoder_udf = udf(encoder, ArrayType(IntegerType()))

现在我们可以汇总 feature 按列分组的列表 label 使用 collect_list() ，并应用我们的 udf 按行：

from pyspark.sql.functions import collect_list, explode

df1 = (df.groupBy("label")
       .agg(collect_list("feature")
            .alias("features"))
       .withColumn("index", 
                   encoder_udf("features")))

因此，可以将 index 列以获取编码值而不是字母：

df1.select("label", explode(df1.index).alias("index")).show()
+-----+-----+
|label|index|
+-----+-----+
|    0|    1|
|    0|    0|
|    0|    0|
|    0|    0|
|    0|    2|
|    1|    0|
|    1|    1|
|    1|    0|
|    1|    0|
+-----+-----+