Spark Scala：无需重复遍历大表添加activeIOsAtSite列的方法

你的当前实现完全不可行，因为activeIOsAtSiteGenerator函数里的count()是Spark的Action操作，会触发全表扫描。每处理原DataFrame的一行，就会执行一次全表扫描，数据量大时会导致Spark提交成百上千次任务，资源直接耗尽，运行效率极低。

高效实现思路 #

要避免重复遍历表，核心是先预处理出所有符合条件的Site ID集合，再将这个集合与原表关联生成目标列，全程只需要遍历原表两次，且可以通过广播优化避免大规模shuffle。

具体实现代码 #

方案一：广播小表+左连接（推荐，适配所有数据规模）

import org.apache.spark.sql.functions.{broadcast, coalesce, lit, upper}

// 1. 提取所有满足条件的Site_siteId，去重后标记为'Y'
val validSiteIdsDf = finvInventoryAllDf
  .filter(
    col("InstalledOffer_installedOfferId").isNotNull &&
    col("InstalledOffer_installedOfferId").notIn("", "null", "NULL") &&
    upper(col("InstalledOffer_standardStatus")) === "ACTIVE" &&
    upper(col("InstalledOffer_applicationSource")) === "CIBASE"
  )
  .select("Site_siteId")
  .distinct()
  .withColumn("activeIOsAtSite", lit("Y"))

// 2. 原表与预处理表左连接，未匹配的填充为'N'
val resultDf = finvInventoryAllDf
  .join(broadcast(validSiteIdsDf), Seq("Site_siteId"), "left_outer")
  .withColumn("activeIOsAtSite", coalesce(col("activeIOsAtSite"), lit("N")))

方案二：广播ID集合+条件判断（适合符合条件的Site ID数量极少的场景）

import org.apache.spark.sql.functions.{lit, upper, when}

// 1. 提取符合条件的Site_siteId集合，广播到所有节点
val validSiteIds = finvInventoryAllDf
  .filter(
    col("InstalledOffer_installedOfferId").isNotNull &&
    col("InstalledOffer_installedOfferId").notIn("", "null", "NULL") &&
    upper(col("InstalledOffer_standardStatus")) === "ACTIVE" &&
    upper(col("InstalledOffer_applicationSource")) === "CIBASE"
  )
  .select("Site_siteId")
  .distinct()
  .as[String]
  .collect()
  .toSet
val broadcastValidSiteIds = spark.sparkContext.broadcast(validSiteIds)

// 2. 直接判断每行的Site ID是否在广播集合中
val resultDf = finvInventoryAllDf
  .withColumn(
    "activeIOsAtSite",
    when(col("Site_siteId").isin(broadcastValidSiteIds.value.toSeq: _*), lit("Y")).otherwise(lit("N"))
  )

方案优势 #

• 仅遍历原表两次：一次提取有效Site ID，一次关联生成目标列
• 广播优化：将小数据集广播到所有Executor节点，避免大规模数据shuffle，大幅提升性能
• 去重操作减少了预处理数据集的大小，进一步降低关联成本

内容的提问来源于stack exchange，提问作者mr.Penguin