Collect_list не работает на частичных агрегатах

Question

Для кадра данных Spark Structured Streaming, collect_list выдает ошибку:

ОШИБКА Исполнитель: Исключение в задаче 0.0 на этапе 11.0 (TID 1024) java.lang.RuntimeException: Collect не может использоваться в частичных агрегациях.

У меня есть следующий код:

df.withColumn("RealValue", expr("(Class, Value)"))
    .groupBy("Code", "Origin", "Destination", "PoS", "Date")
    .agg(collect_list("RealValue").as("ValueSeq"))
    .drop("Value")

Где df - потоковый фрейм данных.

Схема выглядит следующим образом:

val schema = new StructType(Array(
  StructField("Code", StringType),
  StructField("Origin", StringType),
  StructField("Destination", StringType),
  StructField("Date", StringType),
  StructField("PoS", StringType),
  StructField("Class", StringType),
  StructField("Value", StringType)
))

Как мне преодолеть это? Есть ли альтернативный способ сделать это, который будет работать для потокового кадра данных?