这应该可以了，得到一个Array，你可以转换成Map。这里你得到了一个字符串数组，你需要根据逗号分成2个字段，并在Spark Scala Dataframe convert a column of Array of Struct to a column of Map上寻找指导，我投了赞成票。您也可以使用struct代替Array。

import org.apache.spark.sql.functions._
val df = spark.read.textFile("/FileStore/tables/fff1.txt", "/FileStore/tables/fff2.txt")
val df2 = df.flatMap(_.split(" ")).withColumn("filename", input_file_name).groupBy("Value", "filename").count()
val df3 = df2.groupBy("Value").agg(collect_list(array("filename", "count")) as "docs")
df3.show(false)

返回：

+-----+----------------------------------------------------------------------------+
|Value|docs                                                                        |
+-----+----------------------------------------------------------------------------+
|you  |[[dbfs:/FileStore/tables/fff2.txt, 2], [dbfs:/FileStore/tables/fff1.txt, 1]]|
|fine |[[dbfs:/FileStore/tables/fff2.txt, 1], [dbfs:/FileStore/tables/fff1.txt, 1]]|
|how  |[[dbfs:/FileStore/tables/fff2.txt, 1], [dbfs:/FileStore/tables/fff1.txt, 1]]|
|hear |[[dbfs:/FileStore/tables/fff2.txt, 1], [dbfs:/FileStore/tables/fff1.txt, 1]]|
|ok   |[[dbfs:/FileStore/tables/fff2.txt, 1]]                                      |
|have |[[dbfs:/FileStore/tables/fff2.txt, 1]]                                      |
...

为了完整性，要获得Map：

import org.apache.spark.sql.functions._
import org.apache.spark.sql.types._

val arrayToMap = udf[Map[String, Long], Seq[Row]] {
  array => array.map { case Row(key: String, value: Long) => (key, value) }.toMap
}

val df = spark.read.textFile("/FileStore/tables/fff1.txt", "/FileStore/tables/fff2.txt")
val df2 = df.flatMap(_.split(" ")).withColumn("filename", input_file_name).groupBy("Value", "filename").count()
val df3 = df2.groupBy("Value").agg(collect_list(struct("filename", "count")))   

val df4 = df3.withColumn("words", arrayToMap($"collect_list(struct(filename, count))"))
df4.show(false)