小编kru*_*har的帖子

我正在尝试对集合进行一些分析.我有一个示例数据集,如下所示:

orders.json

{"items":[1,2,3,4,5]}
{"items":[1,2,5]}
{"items":[1,3,5]}
{"items":[3,4,5]}

它只是一个字段,它是一个代表ID的数字列表.

这是我试图运行的Spark脚本:

val sparkConf = new SparkConf()
  .setMaster("local[*]")
  .setAppName("Dataframe Test")

val sc = new SparkContext(sparkConf)
val sql = new SQLContext(sc)

val dataframe = sql.read.json("orders.json")

val expanded = dataframe
  .explode[::[Long], Long]("items", "item1")(row => row)
  .explode[::[Long], Long]("items", "item2")(row => row)

val grouped = expanded
  .where(expanded("item1") !== expanded("item2"))
  .groupBy("item1", "item2")
  .count()

val recs = grouped
  .groupBy("item1")

创建expanded并且grouped很好,简而言之,expanded是两个ID在同一原始集中的所有可能的两个ID的列表.grouped过滤掉与自身匹配的ID,然后将所有唯一ID组合在一起并为每个ID生成计数.架构和数据样本grouped是:

root
 |-- item1: long (nullable = true)
 |-- item2: long …