Получение уникальных ключей Map-Column возвращает разные результаты после каждого выполнения

Question

У меня есть фрейм данных pyspark со столбцом MapType (StringType (), FloatType ()), и я бы получил список всех ключей, появляющихся в столбце. Например, имея этот фрейм данных:

+---+--------------------+
| ID|                 map|
+---+--------------------+
|  0|[a -> 3.0, b -> 2...|
|  1|[a -> 1.0, b -> 4...|
|  2|[a -> 6.0, c -> 5...|
|  3|[a -> 6.0, f -> 8...|
|  4|[a -> 2.0, c -> 1...|
|  5|[c -> 1.0, d -> 1...|
|  6|[a -> 4.0, c -> 1...|
|  7|[a -> 2.0, d -> 1...|
|  8|          [a -> 2.0]|
|  9|[e -> 1.0, f -> 1.0]|
| 10|          [g -> 1.0]|
| 11|[e -> 2.0, b -> 3.0]|
+---+--------------------+

Я ожидаю получить следующий список:

['a', 'b', 'c', 'd', 'e', 'f', 'g']

Я уже пробовал

df.select(explode(col('map'))).groupby('key').count().select('key').collect()
df.select(explode(col('map'))).select('key').drop_duplicates().collect()
df.select(explode(col('map'))).select('key').distinct().collect()
df.select(explode(map_keys(col('map')))).select('key').distinct().collect()

...

Но для каждой из этих команд я получаю разные результаты не только для разных команд, но и когда я выполняю одну и ту же команду на одном и том же кадре данных.

Например:

keys_1 = df.select(explode(col('map'))).select('key').drop_duplicates().collect()
keys_1 = [row['key'] for row in keys_1]

И:

keys_2 = df.select(explode(col('map'))).select('key').drop_duplicates().collect()
keys_2 = [row['key'] for row in keys_2]

Тогда довольно часто len (keys_1)! = Len (keys_2).

Мой фрейм данных имеет около 10e7 строк и около 2000 различныхключи для моего столбца карты

Обратите внимание, что на небольшом примере набора данных он работает без проблем, но, к сожалению, довольно сложно найти большой пример набора данных.

Пример кода малого набора данных:

df = spark.createDataFrame([
  (0, {'a':3.0, 'b':2.0, 'c':2.0}),
  (1, {'a':1.0, 'b':4.0, 'd':6.0}),
  (2, {'a':6.0, 'e':5.0, 'c':5.0}),
  (3, {'f':8.0, 'a':6.0, 'g':4.0}),
  (4, {'a':2.0, 'c':1.0, 'd':3.0}),
  (5, {'d':1.0, 'g':5.0, 'c':1.0}),
  (6, {'a':4.0, 'c':1.0, 'f':1.0}),
  (7, {'a':2.0, 'e':2.0, 'd':1.0}),
  (8, {'a':2.0}),
  (9, {'e':1.0, 'f':1.0}),
  (10, {'g':1.0}),
  (11, {'b':3.0, 'e':2.0})
],
  ['ID', 'map']
)
df.select(explode(col('map'))).groupby('key').count().select('key').collect()