Spark Streaming 在实时推荐系统的落地：实时流处理、特征更新与模型服务闭环

/* MD / 富文本中的 .toc（含博客园搬家等嵌套结构）；.toc-box 在侧栏，不受影响 */
#content_views .toc,
/* 编辑器常在目录前后插入空 p（:empty 仍占 20px），一并去掉避免顶空隙 */
#content_views.markdown_views > p:empty:has(+ .toc),
#content_views.markdown_views > .toc + p:empty,
/* 富文本旧版目录标记 */
#content_views.htmledit_views #main-toc,
#content_views.htmledit_views #hr-toc,
#content_views.htmledit_views p[id*=\”-toc\”] {
display: none !important;
}
/* 目录去掉后，紧跟的首个标题不再多出一块上边距 */
#content_views.markdown_views > .toc + h1,
#content_views.markdown_views > .toc + h2,
#content_views.markdown_views > .toc + h3,
#content_views.markdown_views > .toc + h4,
#content_views.markdown_views > .toc + p:empty + h1,
#content_views.markdown_views > .toc + p:empty + h2,
#content_views.markdown_views > .toc + p:empty + h3,
#content_views.markdown_views > .toc + p:empty + h4 {
margin-top: 0 !important;
}

Spark Streaming 在实时推荐系统的落地：实时流处理、特征更新与模型服务闭环

1. 实时推荐系统概述与架构设计

实时推荐系统能够基于用户最新的行为数据即时调整推荐结果，极大提升用户体验与转化率。基于 Spark Streaming 的推荐系统架构需要解决三大核心问题：如何高效处理用户行为流、如何实时更新特征工程、如何将新特征应用到模型服务中。

实时推荐系统的整体架构可以分为数据采集层、实时处理层、特征服务层和模型服务层四个部分。数据采集层负责收集各类用户行为和上下文数据；实时处理层利用 Spark Streaming 对数据进行清洗、聚合和特征计算；特征服务层负责特征的存储、更新和访问；模型服务层则将特征与模型结合生成实时推荐结果。

如图所示，整个系统形成从数据采集到推荐结果输出的闭环。数据采集层通过 Kafka 或 Flume 等工具收集用户点击、浏览、购买等行为数据；实时处理层基于 Spark Streaming 进行数据处理与特征计算；特征服务层负责存储和管理实时特征；模型服务层则结合特征与模型生成最终推荐结果。

2. 用户行为数据流处理

用户行为数据是实时推荐系统的核心输入，需要设计高效的数据采集和处理机制。首先需要在各个应用端埋点采集用户行为事件，然后通过 Kafka 等消息队列传输到后端，最后由 Spark Streaming 进行实时处理。

2.1 数据采集与传输

用户行为数据采集包括客户端埋点和服务端日志两部分。客户端埋点通常需要记录用户ID、时间戳、行为类型、物品ID等关键字段。采集的数据通过 HTTP 接口发送到 Kafka，形成用户行为事件流。

// 定义用户行为事件样例类
case class UserBehavior(userId: String, itemId: String, category: String,
behaviorType: String, timestamp: Long)

// 从Kafka读取用户行为数据流
val kafkaParams = Map[String, Object](
"bootstrap.servers" -> "localhost:9092",
"key.deserializer" -> classOf[StringDeserializer],
"value.deserializer" -> classOf[StringDeserializer],
"group.id" -> "user_behavior_group",
"auto.offset.reset" -> "latest",
"enable.auto.commit" -> (false: java.lang.Boolean)
)

val topics = Array("user_behavior_topic")
val stream = KafkaUtils.createDirectStream[String, String](
streamingContext,
PreferConsistent,
Subscribe[String, String](topics, kafkaParams)
)

// 解析JSON数据为UserBehavior对象
val behaviorStream = stream.map(record => {
val json = JSON.parseObject(record.value())
UserBehavior(
json.getString("userId"),
json.getString("itemId"),
json.getString("category"),
json.getString("behaviorType"),
json.getLong("timestamp")
)
})

2.2 实时数据清洗与聚合

原始用户行为数据通常包含脏数据或不完整信息，需要进行清洗和标准化处理。同时，针对推荐场景，需要对用户行为进行聚合统计，生成用于特征计算的中间结果。

// 数据清洗：过滤无效行为
def isValidBehavior(behavior: UserBehavior): Boolean = {
behavior.userId != null &&
behavior.userId != "" &&
behavior.itemId != null &&
behavior.itemId != "" &&
behavior.timestamp > System.currentTimeMillis() – 30 * 24 * 60 * 60 * 1000L // 过滤30天前的数据
}

val cleanBehaviorStream = behaviorStream.filter(isValidBehavior)

// 按用户分组，统计每个用户的行为次数
val userBehaviorCount = cleanBehaviorStream
.map(behavior => (behavior.userId, 1))
.reduceByKey(_ + _)
.updateStateByKey { (values: Seq[Int], state: Option[Int]) =>
Some(state.getOrElse(0) + values.sum)
}

// 按用户-物品对分组，统计用户对物品的互动行为
val userItemBehavior = cleanBehaviorStream
.map(behavior => ((behavior.userId, behavior.itemId), 1))
.reduceByKey(_ + _)
.updateStateByKey { (values: Seq[Int], state: Option[Int]) =>
Some(state.getOrElse(0) + values.sum)
}

如图所示，用户行为数据处理包括从原始JSON数据到最终特征存储的完整链路。首先进行JSON解析和数据验证，然后通过滑动窗口进行时间窗口处理，接着进行用户行为统计和特征提取，最后将特征存储到Redis等内存数据库中供模型服务使用。

2.3 行为数据存储与查询

处理后的用户行为数据需要高效存储和快速查询，以满足实时推荐系统的低延迟要求。通常使用 Redis 作为热数据存储，HBase 作为持久化存储，两者结合实现高效查询与持久化保障。

// 将用户行为统计结果写入Redis
def writeUserBehaviorToRedis(data: DStream[(String, Int)]): Unit = {
data.foreachRDD { rdd =>
rdd.foreachPartition { partition =>
val jedis = RedisConnectionPool.getResource()
try {
partition.foreach { case (userId, behaviorCount) =>
jedis.hset("user_behavior", userId, behaviorCount.toString)
jedis.expire("user_behavior", 7 * 24 * 60 * 60) // 设置7天过期
}
} finally {
jedis.close()
}
}
}
}

// 将用户-物品交互统计写入Redis
def writeUserItemBehaviorToRedis(data: DStream[((String, String), Int)]): Unit = {
data.foreachRDD { rdd =>
rdd.foreachPartition { partition =>
val jedis = RedisConnectionPool.getResource()
try {
partition.foreach { case ((userId, itemId), interactionCount) =>
val key = s"user_item_interaction:$userId:$itemId"
jedis.set(key, interactionCount.toString)
jedis.expire(key, 30 * 24 * 60 * 60) // 设置30天过期
}
} finally {
jedis.close()
}
}
}
}

3. 实时特征工程与更新

实时特征工程是实时推荐系统的核心环节，它需要根据用户最新的行为数据实时计算特征更新，确保特征能够反映用户的最新兴趣偏好。与离线特征工程不同，实时特征工程强调低延迟、高吞吐和增量更新。

3.1 特征类型与计算策略

实时推荐系统中常用特征包括用户基本特征、物品基本特征、用户行为统计特征和实时交互特征等。不同特征类型的计算策略和更新频率也有所不同。

// 定义各种实时特征的计算策略
object RealTimeFeatures {
// 用户实时兴趣特征
def userInterestFeatures(userBehaviors: Iterable[UserBehavior]): Map[String, Double] = {
val categoryCount = userBehaviors.groupBy(_.category).mapValues(_.size)
val totalCount = userBehaviors.size

categoryCount.map { case (category, count) =>
s"user_interest_$category" -> (count.toDouble / totalCount)
}
}

// 用户行为活跃度特征
def userActivityFeatures(userBehaviors: Iterable[UserBehavior]): Map[String, Double] = {
val behaviorsByType = userBehaviors.groupBy(_.behaviorType)
val totalCount = userBehaviors.size

behaviorsByType.map { case (behaviorType, behaviors) =>
s"user_activity_$behaviorType" -> (behaviors.size.toDouble / totalCount)
}
}

// 物品流行度特征
def itemPopularityFeatures(userBehaviors: Iterable[UserBehavior]): Map[String, Double] = {
val itemCount = userBehaviors.groupBy(_.itemId).mapValues(_.size)
val totalCount = userBehaviors.size

itemCount.map { case (itemId, count) =>
s"item_popularity_$itemId" -> (count.toDouble / totalCount)
}
}
}

3.2 实时特征更新机制

实时特征需要设计高效的更新机制，确保特征能够及时反映用户和物品的最新状态。通常采用增量更新与定期全量更新相结合的策略，在保证数据新鲜度的同时减少计算资源消耗。

// 基于滑动窗口的特征更新
val windowDuration = Minutes(30) // 30分钟滑动窗口
val slideDuration = Seconds(10) // 10秒滑动一次

val windowedBehaviorStream = cleanBehaviorStream
.window(windowDuration, slideDuration)

// 计算用户实时兴趣特征
val userInterestFeatureStream = windowedBehaviorStream
.map(behavior => (behavior.userId, behavior))
.groupByKeyAndWindow(windowDuration, slideDuration)
.map { case (userId, behaviors) =>
val features = RealTimeFeatures.userInterestFeatures(behaviors)
(userId, features)
}

// 计算用户活跃度特征
val userActivityFeatureStream = windowedBehaviorStream
.map(behavior => (behavior.userId, behavior))
.groupByKeyAndWindow(windowDuration, slideDuration)
.map { case (userId, behaviors) =>
val features = RealTimeFeatures.userActivityFeatures(behaviors)
(userId, features)
}

如图所示，实时特征更新机制包括从用户行为流开始，经过滑动窗口处理、特征计算、特征合并、特征过滤、特征时效性检查，最终到特征存储的完整流程。整个系统采用30分钟窗口、10秒滑动的处理机制，支持增量更新和TTL管理，确保特征的新鲜度和存储效率。

3.3 特征存储与访问优化

实时特征的高效存储和快速访问是保障推荐系统响应速度的关键。通常采用多级存储策略，热点特征存储在Redis中，冷门特征存储在HBase中，并通过预加载、缓存等机制优化访问性能。

// 特征写入Redis
def writeFeaturesToRedis(features: DStream[(String, Map[String, Double])]): Unit = {
features.foreachRDD { rdd =>
rdd.foreachPartition { partition =>
val jedis = RedisConnectionPool.getResource()
try {
partition.foreach { case (userId, featureMap) =>
// 使用Hash存储用户特征
jedis.hset("user_features", userId, JSON.toJSONString(featureMap))

// 设置过期时间
jedis.expire("user_features", 24 * 60 * 60) // 24小时过期

// 为特征建立索引，便于快速查询
featureMap.keys.foreach { featureName =>
jedis.sadd("feature_index", featureName)
}
}
} finally {
jedis.close()
}
}
}
}

// 从Redis批量读取用户特征
def batchGetUserFeatures(userIds: List[String]): Map[String, Map[String, Double]] = {
val jedis = RedisConnectionPool.getResource()
try {
val pipeline = jedis.pipelined()
userIds.foreach { userId =>
pipeline.hget("user_features", userId)
}

val results = pipeline.exec().asInstanceOf[java.util.List[String]]
userIds.zip(results).filter { case (_, value) => value != null }.map { case (userId, value) =>
userId -> JSON.parseObject(value, classOf[Map[String, Object]]).mapValues(_.toString.toDouble)
}.toMap
} finally {
jedis.close()
}
}

4. 模型服务与实时推理

实时特征与模型结合生成推荐结果是实时推荐系统的最后一公里，需要设计低延迟、高可用的模型服务架构，同时支持模型的在线更新和A/B测试。

4.1 模型在线服务设计

模型在线服务需要同时支持模型热加载和请求低延迟，通常采用基于Netty的异步服务框架，结合内存缓存和批处理优化实现高性能推理。

// 模型服务接口定义
public interface ModelService {
// 批量预测接口
Map<String, List<ItemScore>> batchPredict(List<String> userIds, List<String> items);

// 单用户预测接口
List<ItemScore> predict(String userId, List<String> items);

// 特征获取接口
Map<String, Double> getUserFeatures(String userId);
}

// 模型服务实现
class ModelServiceImpl implements ModelService {
private final ModelLoader modelLoader;
private final FeatureService featureService;

public ModelServiceImpl(ModelLoader modelLoader, FeatureService featureService) {
this.modelLoader = modelLoader;
this.featureService = featureService;
}

@Override
public List<ItemScore> predict(String userId, List<String> items) {
// 获取用户特征
Map<String, Double> userFeatures = featureService.getUserFeatures(userId);

// 构建预测请求
PredictRequest request = new PredictRequest(userId, items, userFeatures);

// 执行预测
return modelLoader.getModel().predict(request);
}

@Override
public Map<String, List<ItemScore>> batchPredict(List<String> userIds, List<String> items) {
// 批量获取用户特征
Map<String, Map<String, Double>> userFeatures = featureService.batchGetUserFeatures(userIds);

// 批量预测
return modelLoader.getModel().batchPredict(userIds, items, userFeatures);
}
}

4.2 实时特征与模型结合

实时推荐的关键在于将实时获取的特征与模型结合，生成针对用户最新兴趣的推荐结果。需要设计特征提取和模型推理的协同机制，确保特征的新鲜度和模型的一致性。

// 特征提取与模型预测协同
class RealTimeRecommendationService {
private final ModelService modelService;
private final RealTimeFeatureService featureService;

public RealTimeRecommendationService(ModelService modelService,
RealTimeFeatureService featureService) {
this.modelService = modelService;
this.featureService = featureService;
}

// 实时推荐接口
public RecommendationResult recommend(String userId, int numItems) {
// 获取用户实时特征
Map<String, Double> userFeatures = featureService.getRealTimeFeatures(userId);

// 获取候选物品
List<String> candidateItems = getCandidateItems(userId, numItems);

// 预测用户对物品的偏好
List<ItemScore> itemScores = modelService.predict(userId, candidateItems);

// 结合实时特征进行重排序
List<ItemScore> rerankedItems = reRankWithRealTimeFeatures(itemScores, userFeatures);

// 构建推荐结果
return new RecommendationResult(userId, rerankedItems);
}

private List<ItemScore> reRankWithRealTimeFeatures(List<ItemScore> itemScores,
Map<String, Double> userFeatures) {
// 根据实时特征对物品分数进行调整
return itemScores.stream()
.map(itemScore -> {
double realTimeBoost = calculateRealTimeBoost(itemScore.getItemId(), userFeatures);
return new ItemScore(itemScore.getItemId(),
itemScore.getScore() + realTimeBoost);
})
.sorted((a, b) -> Double.compare(b.getScore(), a.getScore()))
.limit(10) // 只返回Top 10
.collect(Collectors.toList());
}

// 计算实时特征带来的分数提升
private double calculateRealTimeBoost(String itemId, Map<String, Double> userFeatures) {
// 实现具体的分数提升计算逻辑
// 可以根据用户实时兴趣与物品属性的匹配度计算
// 这里只是一个示例
return 0.0;
}
}

如图所示，模型服务与推理架构包括从用户请求开始，经过特征服务、模型推理、分数重排序，到结果返回的完整流程。系统同时包含缓存层、模型热加载、A/B测试和监控反馈等支持机制，形成一个完整的闭环系统。

4.3 A/B测试与效果评估

A/B测试是验证实时推荐系统效果的关键手段，需要科学设计实验分组、指标体系和评估方法，确保新模型或新策略的效果可衡量、可对比。

// A/B测试服务实现
class ABTestService {
private final ModelService modelServiceA;
private final ModelService modelServiceB;
private final MetricsCollector metricsCollector;

// 获取推荐结果（根据用户分配不同模型）
public RecommendationResult getRecommendationWithABTest(String userId, int numItems) {
// 根据用户ID决定使用哪个模型
String modelVersion = getUserModelVersion(userId);

// 根据模型版本调用不同的模型服务
if ("A".equals(modelVersion)) {
metricsCollector.recordRequest("model_a");
return modelServiceA.recommend(userId, numItems);
} else {
metricsCollector.recordRequest("model_b");
return modelServiceB.recommend(userId, numItems);
}
}

// 根据用户ID分配模型版本（使用哈希分配确保一致性）
private String getUserModelVersion(String userId) {
int hash = Math.abs(userId.hashCode());
return hash % 2 == 0 ? "A" : "B";
}
}

// 指标收集与评估
class MetricsCollector {
private final MetricsStorage metricsStorage;

// 记录请求指标
public void recordRequest(String modelVersion) {
metricsStorage.incrementCounter("request_count", modelVersion);
}

// 记录点击指标
public void recordClick(String modelVersion, String userId) {
metricsStorage.incrementCounter("click_count", modelVersion);
metricsStorage.recordEvent("user_click", userId, modelVersion);
}

// 记录转化指标
public void recordConversion(String modelVersion, String userId) {
metricsStorage.incrementCounter("conversion_count", modelVersion);
metricsStorage.recordEvent("user_conversion", userId, modelVersion);
}

// 评估指标报告
public EvaluationReport evaluateModelPerformance(String startDate, String endDate) {
// 获取指标数据
Map<String, Long> requestCounts = metricsStorage.getCounters("request_count", startDate, endDate);
Map<String, Long> clickCounts = metricsStorage.getCounters("click_count", startDate, endDate);
Map<String, Long> conversionCounts = metricsStorage.getCounters("conversion_count", startDate, endDate);

// 计算指标
Map<String, Double> ctrRates = calculateCTR(requestCounts, clickCounts);
Map<String, Double> conversionRates = calculateConversionRate(requestCounts, conversionCounts);

// 生成评估报告
return new EvaluationReport(ctrRates, conversionRates);
}

private Map<String, Double> calculateCTR(Map<String, Long> requestCounts,
Map<String, Long> clickCounts) {
Map<String, Double> ctrRates = new HashMap<>();

requestCounts.forEach((model, requestCount) -> {
Long clickCount = clickCounts.getOrDefault(model, 0L);
double ctr = requestCount > 0 ? clickCount.doubleValue() / requestCount : 0.0;
ctrRates.put(model, ctr);
});

return ctrRates;
}
}

5. 完整示例与最佳实践

5.1 完整示例代码

以下是一个完整的实时推荐系统示例代码，展示了从数据采集到特征处理的完整流程：

import org.apache.spark.SparkConf
import org.apache.spark.streaming.{Seconds, StreamingContext}
import org.apache.spark.streaming.kafka010._
import org.apache.kafka.common.serialization.StringDeserializer
import org.apache.spark.streaming.dstream.DStream
import org.json4s.{DefaultFormats, JObject, JString}
import org.json4s.native.JsonMethods._

// 实时推荐系统主程序
object RealTimeRecommendationSystem {

// 用户行为事件样例类
case class UserBehavior(userId: String, itemId: String, category: String,
behaviorType: String, timestamp: Long)

def main(args: Array[String]): Unit = {
// 1. 初始化Spark Streaming环境
val conf = new SparkConf().setAppName("RealTimeRecommendationSystem").setMaster("local[*]")
val ssc = new StreamingContext(conf, Seconds(10))

// 2. 从Kafka读取用户行为数据
val behaviorStream = createKafkaStream(ssc)
val parsedBehaviorStream = parseBehaviorStream(behaviorStream)

// 3. 数据清洗与过滤
val cleanBehaviorStream = filterBehaviorStream(parsedBehaviorStream)

// 4. 实时特征计算
val userFeatureStream = calculateUserFeatures(cleanBehaviorStream)
val itemFeatureStream = calculateItemFeatures(cleanBehaviorStream)

// 5. 实时特征存储
storeRealTimeFeatures(userFeatureStream, itemFeatureStream)

// 6. 启动流处理
ssc.start()
ssc.awaitTermination()
}

// 创建Kafka数据流
private def createKafkaStream(ssc: StreamingContext): DStream[(String, String)] = {
val kafkaParams = Map[String, Object](
"bootstrap.servers" -> "localhost:9092",
"key.deserializer" -> classOf[StringDeserializer],
"value.deserializer" -> classOf[StringDeserializer],
"group.id" -> "realtime_recommendation",
"auto.offset.reset" -> "latest",
"enable.auto.commit" -> (false: java.lang.Boolean)
)

val topics = Array("user_behavior_topic")
KafkaUtils.createDirectStream[String, String](
ssc,
PreferConsistent,
Subscribe[String, String](topics, kafkaParams)
)
}

// 解析行为数据流
private def parseBehaviorStream(behaviorStream: DStream[(String, String)]): DStream[UserBehavior] = {
implicit val formats = DefaultFormats

behaviorStream.map(record => {
val json = parse(record._2).asInstanceOf[JObject]
json.extract[UserBehavior]
})
}

// 过滤无效行为数据
private def filterBehaviorStream(behaviorStream: DStream[UserBehavior]): DStream[UserBehavior] = {
behaviorStream.filter(behavior => {
behavior.userId != null &&
behavior.userId != "" &&
behavior.itemId != null &&
behavior.itemId != "" &&
behavior.timestamp > System.currentTimeMillis() – 30 * 24 * 60 * 60 * 1000L
})
}

// 计算用户特征
private def calculateUserFeatures(behaviorStream: DStream[UserBehavior]): DStream[(String, Map[String, Double])] = {
val windowDuration = Seconds(30 * 60) // 30分钟窗口
val slideDuration = Seconds(10) // 10秒滑动

behaviorStream
.window(windowDuration, slideDuration)
.map(behavior => (behavior.userId, behavior))
.groupByKeyAndWindow(windowDuration, slideDuration)
.map { case (userId, behaviors) =>
// 计算用户兴趣特征
val categoryCount = behaviors.groupBy(_.category).mapValues(_.size)
val totalCount = behaviors.size

val interestFeatures = categoryCount.map { case (category, count) =>
s"user_interest_$category" -> (count.toDouble / totalCount)
}.toMap

// 计算用户活跃度特征
val behaviorCount = behaviors.groupBy(_.behaviorType).mapValues(_.size)
val activityFeatures = behaviorCount.map { case (behaviorType, count) =>
s"user_activity_$behaviorType" -> (count.toDouble / totalCount)
}.toMap

// 合并所有用户特征
val allFeatures = interestFeatures ++ activityFeatures

(userId, allFeatures)
}
}

// 计算物品特征
private def calculateItemFeatures(behaviorStream: DStream[UserBehavior]): DStream[(String, Map[String, Double])] = {
val windowDuration = Seconds(30 * 60) // 30分钟窗口
val slideDuration = Seconds(10) // 10秒滑动

behaviorStream
.window(windowDuration, slideDuration)
.map(behavior => (behavior.itemId, behavior))
.groupByKeyAndWindow(windowDuration, slideDuration)
.map { case (itemId, behaviors) =>
// 计算物品流行度特征
val totalCount = behaviors.size

val popularityFeatures = behaviors.groupBy(_.behaviorType).map { case (behaviorType, items) =>
s"item_popularity_$behaviorType" -> (items.size.toDouble / totalCount)
}.toMap

// 计算物品点击率特征
val clickCount = behaviors.count(_.behaviorType == "click")
val viewCount = behaviors.count(_.behaviorType == "view")

val ctrFeatures = Map(
"item_ctr" -> (if (viewCount > 0) clickCount.toDouble / viewCount else 0.0)
)

// 合并所有物品特征
val allFeatures = popularityFeatures ++ ctrFeatures

(itemId, allFeatures)
}
}

// 存储实时特征
private def storeRealTimeFeatures(userFeatureStream: DStream[(String, Map[String, Double])],
itemFeatureStream: DStream[(String, Map[String, Double])]): Unit = {
// 存储用户特征
userFeatureStream.foreachRDD { rdd =>
rdd.foreachPartition { partition =>
val jedis = RedisConnectionPool.getResource()
try {
partition.foreach { case (userId, features) =>
jedis.hset("user_features", userId, features.toString())
jedis.expire("user_features", 24 * 60 * 60) // 24小时过期
}
} finally {
jedis.close()
}
}
}

// 存储物品特征
itemFeatureStream.foreachRDD { rdd =>
rdd.foreachPartition { partition =>
val jedis = RedisConnectionPool.getResource()
try {
partition.foreach { case (itemId, features) =>
jedis.hset("item_features", itemId, features.toString())
jedis.expire("item_features", 24 * 60 * 60) // 24小时过期
}
} finally {
jedis.close()
}
}
}
}
}

5.2 最佳实践与注意事项

通过以上步骤和最佳实践，我们可以构建一个高效稳定的基于Spark Streaming的实时推荐系统，为用户提供更加精准和个性化的推荐服务。