Spark分析不同年份共同前往国家的人群及出行信息方案咨询

问题分析 #

你遇到的核心问题是用FPGrowth直接处理全量数据时，丢失了年份和国家这两个关键维度的关联。FPGrowth本身是用来挖掘频繁项集的，但我们需要的是同一年份下，人群组合共同前往的国家集合以及对应的次数，完全不需要用到协同过滤（协同过滤更多用于用户-物品的推荐场景）。更合适的思路是先按年份拆分数据，再针对每个年份内的人群-国家关系做关联分析，既可以直接生成所有有效人群组合，也可以结合FPGrowth挖掘高频人群组合后匹配对应国家信息。

具体实现步骤 #

我们可以通过三步完成需求，先聚合基础数据，再生成人群组合，最后计算共同国家和次数：

1. 按年份+人群分组，得到每个人在对应年份去过的国家集合
2. 按年份生成该年份内所有2人及以上的非重复人群组合
3. 计算每个组合的共同国家集合，次数即为共同国家的数量

下面是完整的Scala代码实现：

val spark = SparkSession.builder().master("local[1,1]").config("spark.sql.shuffle.partitions", "1").config("spark.sql.crossJoin.enabled","true").getOrCreate()
spark.sparkContext.setLogLevel("error")
import spark.implicits._
import org.apache.spark.sql.functions._

// 加载原始数据
val df=Seq(
 ("tom","America","2019"),
 ("jim","America","2019"),
 ("jack","America","2019"),
 ("tom","Russia","2019"),
 ("jim","Russia","2019"),
 ("jack","Russia","2019"),
 ("alex","Russia","2019"),
 ("tom","America","2018"),
 ("jim","America","2018"),
 ("tom","Germany","2018"),
 ("jim","England","2018")
).toDF("person","country","year")

// 步骤1：按年份和人分组，得到每个人在对应年份的国家集合
val personCountryByYear = df.groupBy("year", "person")
  .agg(collect_set("country").alias("countries"))
  .orderBy("year", "person")

// 步骤2：定义UDF处理单年份内的人群组合与共同国家计算
def processYearGroup(personCountryList: Seq[(String, Seq[String])]): Seq[(Seq[String], Seq[String], Int)] = {
  // 生成所有2人及以上的非重复组合
  val allValidCombinations = (2 to personCountryList.size).flatMap { groupSize =>
    personCountryList.combinations(groupSize).map(_.map(_._1))
  }
  // 对每个组合计算共同国家与次数
  allValidCombinations.map { persons =>
    val countrySets = persons.map(p => personCountryList.find(_._1 == p).get._2.toSet)
    val commonCountries = countrySets.reduce(_ intersect _).toSeq.sorted
    val times = commonCountries.size
    (persons.sorted, commonCountries, times)
  }.filter(_._3 > 0) // 过滤无共同国家的无效组合
}

val processYearUDF = udf(processYearGroup _)

// 步骤3：应用UDF并整理成目标格式
val result = personCountryByYear.groupBy("year")
  .agg(collect_list(struct("person", "countries")).alias("person_country_list"))
  .withColumn("items", processYearUDF(col("person_country_list")))
  .select("year", "items")
  .withColumn("items", explode(col("items")))
  .select(
    col("year"),
    col("items._1").alias("persons"),
    col("items._2").alias("common"),
    col("items._3").alias("times")
  )
  .groupBy("year")
  .agg(collect_list(struct("persons", "common", "times")).alias("items"))
  .orderBy("year")

// 输出为目标JSON格式
result.toJSON.show(false)

针对大数据量的优化（用FPGrowth挖掘高频人群组合） #

如果你的数据量很大，不需要所有人群组合，只关注频繁共同出行的人群，可以在每个年份内用FPGrowth挖掘频繁项集，再匹配共同国家：

import org.apache.spark.ml.fpm.FPGrowth

// 按年份拆分数据
val yearList = personCountryByYear.select("year").distinct().collect().map(_.getString(0))

val finalResultDFs = yearList.map { year =>
  // 生成当前年份的人群列表，适配FPGrowth输入格式
  val transactionDF = personCountryByYear.filter(col("year") === year)
    .select("person").map(row => Array(row.getString(0))).toDF("items")
  
  // 初始化FPGrowth，调整minSupport控制频繁项集的阈值
  val fpgrowth = new FPGrowth().setItemsCol("items").setMinSupport(0.1)
  val model = fpgrowth.fit(transactionDF)
  
  // 筛选长度>=2的频繁人群组合
  val frequentGroups = model.freqItemsets.filter(col("items").size >= 2)
    .select("items").collect().map(_.getAs[Seq[String]](0))
  
  // 构建当前年份的人群-国家映射
  val personCountryMap = personCountryByYear.filter(col("year") === year)
    .rdd.map(row => (row.getString(0), row.getAs[Seq[String]](1))).collectAsMap()
  
  // 计算每个高频组合的共同国家与次数
  val items = frequentGroups.map { persons =>
    val countrySets = persons.map(p => personCountryMap(p).toSet)
    val commonCountries = countrySets.reduce(_ intersect _).toSeq.sorted
    val times = commonCountries.size
    (persons.sorted, commonCountries, times)
  }.filter(_._3 > 0)
  
  // 转换为DataFrame并标记年份
  spark.createDataFrame(items).toDF("persons", "common", "times")
    .withColumn("year", lit(year))
}

// 合并所有年份结果并整理格式
val finalResult = finalResultDFs.reduce(_ union _)
  .groupBy("year")
  .agg(collect_list(struct("persons", "common", "times")).alias("items"))
  .orderBy("year")

finalResult.toJSON.show(false)

内容的提问来源于stack exchange，提问作者eascape