映射 Spark Row 中数组的每个值

Question

我有一个json数据集，格式如下，每行一个条目。

 { "sales_person_name" : "John", "products" : ["apple", "mango", "guava"]}
 { "sales_person_name" : "Tom", "products" : ["mango", "orange"]}
 { "sales_person_name" : "John", "products" : ["apple", "banana"]}
 { "sales_person_name" : "Steve", "products" : ["apple", "mango"]}
 { "sales_person_name" : "Tom", "products" : ["mango", "guava"]}

我想知道谁卖的芒果最多等等。 因此，我想将文件加载到数据帧并为每个事务的数组中的每个产品值发出一个（键，值）对（产品，名称）。

var df = spark.read.json("s3n://sales-data.json")
df.printSchema()
root
 |-- sales_person_name: string (nullable = true)
 |-- products: array (nullable = true)

var nameProductsMap = df.select("sales_person_name",  "products").show()
+-----------------+--------------------+
|sales_person_name|   products         |
+-----------------+--------------------+
|             John|[mango, apple,...   |
|              Tom|[mango, orange,...  |
|             John|[apple, banana...   | 

var resultMap = df.select("products", "sales_person_name")
                  .map(r => (r(1), r(0)))
                  .show()  //This is where I am stuck.

我无法找出正确的方法来爆炸（）行（0）并使用行（1）值发出一次所有值。任何人都可以提出一种方法。谢谢！

预期输出：

Mango : John(4), Tom(2), Greg(1)... 
Banana: Tom(5), John(2), ...

Answer 1

val exploded = df.explode("products", "product") { a: mutable.WrappedArray[String] => a }
val result = exploded.drop("products")
result.show()

打印：

+-----------------+-------+
|sales_person_name|product|
+-----------------+-------+
|             John|  apple|
|             John|  mango|
|             John|  guava|
|              Tom|  mango|
|              Tom| orange|
|             John|  apple|
|             John| banana|
|            Steve|  apple|
|            Steve|  mango|
|              Tom|  mango|
|              Tom|  guava|
+-----------------+-------+

Answer 2

更新

下面的代码应该可以工作

import org.apache.spark.sql.functions.explode
import scala.collection.mutable

val resultMap = df.select(explode($"products"), $"sales_person_name")


def counter(l: TraversableOnce[Any]) = {
  val temp = mutable.Map[Any, Int]()
    for (i <- l) {
      if(temp.contains(i)) temp(i) += 1
      else temp(i) = 1
    }
  temp
}

resultsMap.map(x => (x(0), Array(x(1)))).
           reduceByKey(_ ++ _).
           map { case (x,y) => (x, counter(y).toArray) }

结果输出：Array((banana,Array((John,1))), (guava,Array((Tom,1), (John,1))), (orange,Array((Tom,1))), (apple,Array((Steve,1), (John,2))), (mango,Array((Tom,2), (Steve,1), (John,1))))