Искра структурированная потоковая передача, объединяющая совокупный фрейм данных с фреймом данных

У меня есть потоковый фрейм данных, который может выглядеть примерно так:

+--------------------+--------------------+
|               owner|              fruits|
+--------------------+--------------------+
|Brian                | apple|
Brian                | pear |
Brian                | date|
Brian                | avocado|
Bob                | avocado|
Bob                | apple|
........
+--------------------+--------------------+

Я выполнил groupBy, agg collect_list, чтобы навести порядок.

val myFarmDF = farmDF.withWatermark("timeStamp", "1 seconds").groupBy("fruits").agg(collect_list(col("fruits")) as "fruitsA")

на выходе получается одна строка для каждого владельца и массив каждого фрукта. Теперь я хотел бы присоединить этот очищенный массив к исходному фрейму данных потоковой передачи, отбросив столбец фруктов и просто получив столбец плод

val joinedDF = farmDF.join(myFarmDF, "owner").drop("fruits")

это, кажется, работает в моей голове, но Спарк, похоже, не согласен.

Я получаю

Failure when resolving conflicting references in Join:
'Join Inner
...
+- AnalysisBarrier
      +- Aggregate [name#17], [name#17, collect_list(fruits#61, 0, 0) AS fruitA#142]

Когда я превращаю все в статический фрейм данных, он работает нормально. Разве это невозможно в контексте потоковой передачи?