基于Spark Structured Streaming构建推荐系统：如何增量查询数据库数据至流数据集？

针对你用Spark Structured Streaming构建推荐系统的需求——消费Kafka的用户搜索请求，关联数据库中的用户历史订单（不全量加载）来生成推荐，我整理了几个实用的方案，结合场景给你分析：

方案1：用foreachBatch实现微批级批量查询（最常用） #

这个思路是只查询当前微批涉及到的用户的历史订单，避免全量加载数据库数据，非常适合订单数据更新不那么频繁的场景。

核心步骤：

• 在每个微批处理时，先提取当前批次里的所有唯一用户ID
• 用这些用户ID作为过滤条件，批量查询数据库中的对应订单数据
• 将查询到的订单数据和当前微批的搜索数据关联，计算推荐

代码示例（Scala）：

import org.apache.spark.sql.SparkSession
import org.apache.spark.sql.functions.{get_json_object, collect_list}

val spark = SparkSession.builder()
  .appName("SearchRecommendation")
  .getOrCreate()

// 读取Kafka的搜索流数据
val streamDF = spark.readStream
  .format("kafka")
  .option("kafka.bootstrap.servers", "host:port")
  .option("subscribe", "user-searches")
  .load()
  .selectExpr("CAST(value AS STRING)")
  // 解析消息为user_id和search_keyword
  .select(
    get_json_object($"value", "$.user_id").alias("user_id"),
    get_json_object($"value", "$.search_keyword").alias("search_keyword")
  )

// 用foreachBatch处理每个微批
streamDF.writeStream.foreachBatch { (batchDF: DataFrame, batchId: Long) =>
  // 提取当前微批的唯一用户ID
  val userIds = batchDF.select("user_id").distinct().as[String].collect()
  
  if (userIds.nonEmpty) {
    // 构建批量查询的SQL，注意处理SQL注入风险（字符串类型ID加引号）
    val userIdsStr = userIds.map(id => s"'$id'").mkString(",")
    val ordersQuery = s"""
      SELECT user_id, item_id, order_time 
      FROM orders 
      WHERE user_id IN ($userIdsStr)
    """
    
    // 批量查询数据库订单
    val ordersDF = spark.read
      .format("jdbc")
      .option("url", "jdbc:mysql://db-host:3306/your-db")
      .option("dbtable", s"($ordersQuery) AS user_orders")
      .option("user", "db-user")
      .option("password", "db-pass")
      .option("driver", "com.mysql.cj.jdbc.Driver")
      .load()
    
    // 关联搜索数据和订单数据，加入推荐逻辑
    val recommendationDF = batchDF.join(ordersDF, Seq("user_id"), "left_outer")
      .groupBy("user_id", "search_keyword")
      .agg(collect_list("item_id").alias("history_items"))
      // 替换成你的推荐算法，比如基于关键词匹配历史订单的商品
      // .withColumn("recommendations", yourRecommendationUDF($"search_keyword", $"history_items"))
    
    // 输出推荐结果到Kafka或其他存储
    recommendationDF.write
      .format("kafka")
      .option("kafka.bootstrap.servers", "host:port")
      .option("topic", "user-recommendations")
      .save()
  }
}.start().awaitTermination()

注意事项：

• 避免SQL注入：如果用户ID是字符串类型，一定要加引号转义；数字类型也要注意格式规范。
• 分批次查询：如果当前微批用户ID过多（比如上万条），要拆分多个IN子句分批查询，避免SQL语句过长导致数据库报错。
• 连接复用：可以在Executor级别初始化数据库连接池（比如用静态变量初始化HikariCP），减少连接创建开销。

方案2：基于CDC+Stateful Processing维护实时用户订单状态 #

如果你的订单数据是实时更新的（比如用户下单后立刻要纳入推荐），可以用**Change Data Capture（CDC）**把数据库的订单变更同步到Kafka，然后用Spark的状态流处理来维护每个用户的历史订单，这样流处理时直接从状态中读取，不用查数据库。

核心步骤：

1. 用Debezium等工具捕获数据库的订单新增/更新/删除事件，发送到Kafka Topic
2. 读取这个订单流，用flatMapGroupsWithState维护每个用户的订单列表（支持状态过期）
3. 将搜索流和订单状态流关联，生成推荐

代码示例（Scala）：

import org.apache.spark.sql.functions.from_json
import org.apache.spark.sql.streaming.{GroupState, GroupStateTimeout, OutputMode}

// 定义订单状态和事件的样例类
case class OrderState(items: Seq[String])
case class OrderEvent(user_id: String, item_id: String, event_type: String, order_time: String)

// 读取订单CDC流
val orderStreamDF = spark.readStream
  .format("kafka")
  .option("kafka.bootstrap.servers", "host:port")
  .option("subscribe", "orders-cdc")
  .load()
  .selectExpr("CAST(value AS STRING)")
  .select(from_json($"value", schemaOf[OrderEvent]).alias("event"))
  .select("event.*")
  .withColumn("order_time", $"order_time".cast("timestamp"))

// 维护用户订单状态
val userOrderStateStream = orderStreamDF
  .withWatermark("order_time", "1 day") // 设置水位线，用于状态过期
  .groupByKey(_.user_id)
  .flatMapGroupsWithState(OutputMode.Update(), GroupStateTimeout.ProcessingTimeTimeout()) {
    case (userId: String, events: Iterator[OrderEvent], state: GroupState[OrderState]) =>
      // 处理新增/删除事件，更新状态
      val currentItems = state.getOption.map(_.items).getOrElse(Seq.empty)
      val updatedItems = events.foldLeft(currentItems) { (items, event) =>
        event.event_type match {
          case "INSERT" => items :+ event.item_id
          case "DELETE" => items.filter(_ != event.item_id)
          case _ => items
        }
      }
      state.update(OrderState(updatedItems))
      state.setTimeoutDuration("7 days") // 设置状态7天过期，避免内存溢出
      Iterator((userId, updatedItems))
  }

// 关联搜索流和订单状态流
val recommendationStream = streamDF.join(
  userOrderStateStream.toDF("user_id", "history_items"),
  Seq("user_id"),
  "left_outer"
).withColumn("recommendations", yourRecommendationUDF($"search_keyword", $"history_items"))

// 输出结果
recommendationStream.writeStream
  .format("kafka")
  .option("topic", "user-recommendations")
  .start()
  .awaitTermination()

这个方案的优势是实时性高，订单变更立刻能反映到推荐中；缺点是需要维护CDC管道，状态管理要注意内存占用（设置合理的过期时间）。

方案3：自定义UDF+连接池实现按需单条查询 #

如果你的微批中用户数量很少，或者需要针对每个用户做实时查询，可以写一个自定义UDF，用连接池来复用数据库连接，根据用户ID查询历史订单。

代码示例（Scala）：

import com.zaxxer.hikari.{HikariConfig, HikariDataSource}
import org.apache.spark.sql.functions.udf

// 初始化连接池（每个Executor初始化一次）
object DBConnectionPool {
  private val config = new HikariConfig()
  config.setJdbcUrl("jdbc:mysql://db-host:3306/your-db")
  config.setUsername("db-user")
  config.setPassword("db-pass")
  config.setDriverClassName("com.mysql.cj.jdbc.Driver")
  config.setMaximumPoolSize(10) // 根据Executor资源调整
  
  val dataSource = new HikariDataSource(config)
}

// 自定义UDF：根据用户ID查询历史订单
val getHistoryItems = udf((userId: String) => {
  var conn = null
  var stmt = null
  var rs = null
  val items = collection.mutable.ListBuffer[String]()
  try {
    conn = DBConnectionPool.dataSource.getConnection()
    stmt = conn.prepareStatement("SELECT item_id FROM orders WHERE user_id = ?")
    stmt.setString(1, userId)
    rs = stmt.executeQuery()
    while (rs.next()) {
      items += rs.getString("item_id")
    }
  } catch {
    case e: Exception => e.printStackTrace()
  } finally {
    rs?.close()
    stmt?.close()
    conn?.close()
  }
  items.toSeq
})

// 在流中使用UDF生成推荐
val recommendationDF = streamDF.withColumn("history_items", getHistoryItems($"user_id"))
  .withColumn("recommendations", yourRecommendationUDF($"search_keyword", $"history_items"))

// 输出结果
recommendationDF.writeStream
  .format("kafka")
  .option("topic", "user-recommendations")
  .start()
  .awaitTermination()

注意事项：

• 连接池初始化：要确保每个Executor只初始化一次连接池，避免资源浪费。
• 性能问题：单条查询的效率不如批量查询，适合用户量小的场景；如果用户量大，会给数据库带来很大压力。
• 异常处理：一定要在UDF中加入异常捕获和资源释放逻辑，避免连接泄漏。

总结选择建议 #

• 如果订单数据更新不频繁，优先选方案1（foreachBatch批量查询），实现简单，资源开销小。
• 如果订单数据需要实时同步到推荐中，选方案2（CDC+Stateful），实时性最高。
• 如果微批用户量极少，且需要实时查询，选方案3（自定义UDF），灵活性高但性能一般。

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