【免费下载链接】recommenders
Best Practices on Recommendation Systems
项目地址：
https://gitcode.com/gh_mirrors/re/recommenders
点击查看 免费下载
导读：SARplus 是微软 Recommenders 开源仓库中基于 Apache Spark 的 SAR（Simple Algorithm for Recommendation，简单推荐算法）高效实现，通过 PySpark 驱动、C++ 内存映射加速推理。本文以仓库内 contrib/sarplus/python/README.md 为骨架，系统讲解 SAR 算法原理、SARplus 的核心优化手段、参数配置、三种部署环境（本地/PySpark Shell、Databricks、Azure Synapse）的安装与缓存挂载方案，并结合源码给出底层实现依据。读完本文，你将能在 Spark 集群上完成 SAR 模型的训练与 Top-K 推荐。
一、SAR 算法与 SARplus 概览
SAR（Simple Algorithm for Recommendation）是一种基于**邻域（neighborhood）**的个性化推荐算法，核心思想建立在用户历史交易行为之上：
- Affinity（亲和度）：用户过去与某物品产生过交互（购买、点击、评分等），就认为该用户对该物品存在亲和度；
- Similarity（相似度）：两个物品相似，是指与其中一个物品交互过的用户，也很可能与另一个物品发生交互——即"共同交互用户"越多，物品越相似；
- 推荐逻辑：SAR 向用户推荐与其已有亲和度物品最相似的物品，本质上是"用户已喜欢的物品 + 物品间相似度 → 候选排序"。
该算法原理在 contrib/sarplus/python/README.md 中有精确定义，且与仓库中单机版 SAR（recommenders/models/sar/sar_singlenode.py）保持一致的数学基础。
SARplus 是这一算法在 Spark 上的高效实现，仓库 contrib/sarplus/README.md 明确列出了三大特性：
仓库还给出一个官方基准参考：在 Databricks 8 个 worker（Azure Standard DS3 v2，4 核机器）环境下，处理约 250 万用户、3.5 万物品、1 亿条评分数据，总运行时长约 1.3 小时。该数据仅为仓库 README 声明的环境示例，实际性能取决于集群规模与数据分布。
二、核心优化原理：从相似度矩阵到 C++ 内存映射
SARplus 之所以能支撑亿级评分数据，关键在于一组针对 Top-K 推荐链路的深度优化（详见 contrib/sarplus/README.md 的 "Top-K Recommendation Optimization" 一节）：
| 物品 ID 连续化索引 | 将字符串等原始物品 ID 映射为连续整数索引，压缩存储并简化内存访问，对应 _recommend_k_items_fast 中 item_mapping 表的构建（SARPlus.py） |
| 按 C++ 组件所需格式直接导出相似度矩阵 | 相似度矩阵先被映射到索引空间，再通过 Scala 自定义数据源 com.microsoft.sarplus 以与 C++ 内存布局完全一致的方式落盘，省去推理时的解析开销，见 SARPlus.py 与 DefaultSource.scala |
| 共享只读内存映射 | C++ 端通过 mmap(MAP_SHARED) 把 .sar 缓存文件映射进进程地址空间（pysarplus.cpp），同一 worker 上多个 Python executor 复用同一份物理内存 |
| 按用户分区测试输入 | 测试用户与历史已见物品按用户分区，便于水平扩展预测计算，pred_input.repartition(n_user_prediction_partitions, col_user) 即此逻辑（SARPlus.py） |
| 尽可能在 PySpark 中完成计算 | 只把必须贴近内存数据的 Top-K 排序下沉到 C++，其余（索引映射、亲和度提取、结果回查）全部留在 PySpark，实现简单且易维护 |
| Top-K 计算的三个技巧 | ① 反向 Join：将用户已见物品与相关物品"反接"，先聚合候选得分再排序；② 内存中始终只保留 Top-K 个候选项；③ 用户已见物品与相关物品的 Join 采用二分查找（binary search）加速（pysarplus.cpp 的 join_prod_sum） |
从源码结构看，整条快速推理链路可以概括为：PySpark 准备索引映射与亲和度 → Scala 数据源写出特定二进制布局 → Python UDF 内实例化 C++ 模型 → C++ 内存映射读入矩阵并完成 Top-K 排序 → 结果映射回原始物品 ID。
其中 SARModel（SARModel.py）以进程级单例方式持有 C++ 模型句柄：同一路径只加载一次 .sar 文件（类级缓存 __model），多个批次预测共享同一内存映射，这正是"每 worker 仅缓存一次"的内存优化落点。C++ 端 predict 内部通过最小堆（std::priority_queue）维护 Top-K 候选，并使用 lower_bound 二分查找完成两个有序序列的归并求积（pysarplus.cpp）。
三、Python API 与核心参数详解
3.1 快速上手示例
以 contrib/sarplus/python/README.md 中的最小示例为基线：
from pysarplus import SARPlus
# spark dataframe with user/item/rating/optional timestamp tuples
train_df = spark.createDataFrame(
[(1, 1, 1), (1, 2, 1), (2, 1, 1), (3, 1, 1), (3, 3, 1)],
["user_id", "item_id", "rating"]
)
# spark dataframe with user/item tuples
test_df = spark.createDataFrame(
[(1, 1, 1), (3, 3, 1)],
["user_id", "item_id", "rating"],
)
model = SARPlus(
spark,
col_user="user_id",
col_item="item_id",
col_rating="rating",
col_timestamp="timestamp",
similarity_type="jaccard",
)
model.fit(train_df)
model.recommend_k_items(test_df, top_k=3, remove_seen=False).show()
注意：若想在训练与推理时启用时间衰减或按最新评分去重，col_timestamp 对应的列必须存在；当数据不含时间戳列且 timedecay_formula=False 时，fit 会跳过按时间去重的分支直接使用原始数据（见 SARPlus.py）。
3.2 构造参数完整说明
SARPlus.__init__ 的全部参数及默认值定义在 SARPlus.py：
| spark | （必填） | pyspark.sql.SparkSession 实例 |
| col_user | userID | 用户列名 |
| col_item | itemID | 物品列名 |
| col_rating | rating | 评分列名 |
| col_timestamp | timestamp | 时间戳列名（秒级 Epoch 时间） |
| table_prefix | "" | 训练/中间临时表名的前缀，用于多模型并存避免表名冲突 |
| similarity_type | jaccard | 相似度类型，可选 cooccurrence / jaccard / lift，非法取值会抛出 ValueError（SARPlus.py） |
| time_decay_coefficient | 30 | 评分衰减半衰期（天）：评分经过该天数后衰减一半，公式中作为分母使用；置 0 可关闭时间衰减 |
| time_now | None | 时间衰减计算的当前时间；为 None 时自动取训练数据时间戳最大值（SARPlus.py） |
| timedecay_formula | False | 是否启用时间衰减公式；开启后按 SUM(rating * POW(2, (timestamp – time_now) / time_decay_half_life)) 对用户-物品对聚合（SARPlus.py） |
| threshold | 1 | 物品-物品共现次数低于该阈值的边被剪除；源码断言 threshold > 0 |
| cache_path | None | 本地缓存目录。指定后 recommend_k_items(use_cache=True) 走 C++ 快速预测路径 |
3.3 相似度计算方式
fit 过程在 Spark SQL 中依次完成（SARPlus.py）：
- cooccurrence：直接使用共现次数 value；
- jaccard：value / (m1.margin + m2.margin – value)；
- lift：value / (m1.margin * m2.margin)；
从仓库单机版实现（recommenders/models/sar/sar_singlenode.py）可以印证，这三种相似度与 SAR 标准定义保持一致，只是 SARplus 完全以 Spark SQL 表达、可水平扩展。
3.4 推理参数说明
recommend_k_items(test, top_k=10, remove_seen=True, use_cache=False, n_user_prediction_partitions=200)（SARPlus.py）：
| test | （必填） | 测试集 Spark DataFrame，需包含用户与物品列 |
| top_k | 10 | 每个用户返回的推荐条目数 |
| remove_seen | True | 是否从推荐结果中剔除用户历史已见物品；注意：慢速路径（use_cache=False）当前对 remove_seen=True 会抛出 NotImplementedError（源码中为 ValueError("Not implemented")，见 SARPlus.py），因此慢速路径需使用 remove_seen=False，或直接采用快速路径 |
| use_cache | False | 是否使用 C++ 快速预测；为 True 但未指定 cache_path 时抛 ValueError |
| n_user_prediction_partitions | 200 | 预测阶段的用户分区数，控制推理并行度 |
另外，SARPlus 还提供两个辅助方法：get_topk_most_similar_users(test, user, top_k=10) 计算与目标用户最相似的 Top-K 用户（基于物品向量的内积相似度），get_popularity_based_topk(top_k=10, items=True) 返回全量最热门的 Top-K 物品（基于频率），后者在 items=False 时同样未实现（SARPlus.py）。
四、安装与三种运行环境配置
SARplus 由两个包组成，缺一不可：
- pysarplus（PyPI 包）：Python API + pybind11 C++ 扩展，源码见 contrib/sarplus/python/setup.py，依赖 numpy、pandas、pyarrow>=1.0.0、pybind11>=2.2、pyspark>=3.0.0，支持 Python 3.6～3.11；
- sarplus（Maven 中央仓库包）：Scala 端数据源/格式化器（com.microsoft.sarplus:sarplus_2.12）；若运行在 Spark 3.2+，需改用 com.microsoft.sarplus:sarplus-spark-3-2-plus_2.12——原因是 Spark 3.2 对 OutputWriter 增加了 path() 方法这一破坏性变更，详见 DEVELOPMENT.md 的 "Notes for Spark 3.x" 一节。
以仓库 README 使用的版本 0.6.6 为例：
4.1 Jupyter Notebook / 本地环境
在调用 SARplus 之前插入如下单元格，通过 PYSPARK_SUBMIT_ARGS 注入 Maven 坐标并完成 Spark 会话配置：
import os
SARPLUS_MVN_COORDINATE = "com.microsoft.sarplus:sarplus_2.12:0.6.6"
SUBMIT_ARGS = f"–packages {SARPLUS_MVN_COORDINATE} pyspark-shell"
os.environ["PYSPARK_SUBMIT_ARGS"] = SUBMIT_ARGS
from pyspark.sql import SparkSession
spark = (
SparkSession.builder.appName("sample")
.master("local[*]")
.config("memory", "1G")
.config("spark.sql.shuffle.partitions", "1")
.config("spark.sql.crossJoin.enabled", True)
.config("spark.sql.sources.default", "parquet")
.config("spark.sql.legacy.createHiveTableByDefault", True)
.config("spark.ui.enabled", False)
.getOrCreate()
)
这些配置中，spark.sql.crossJoin.enabled=true 用于启用相似度矩阵计算所需的笛卡尔连接，spark.sql.sources.default=parquet 设置默认存储格式为 parquet，spark.sql.legacy.createHiveTableByDefault=true 是 Spark 3.x 下 saveAsTable 正常工作的前提（DEVELOPMENT.md 明确要求这三项配置）。
4.2 PySpark Shell
SARPLUS_MVN_COORDINATE="com.microsoft.sarplus:sarplus_2.12:0.6.6"
# Install pysarplus
pip install pysarplus
# Specify sarplus maven coordinate and configure Spark environment
pyspark –packages "${SARPLUS_MVN_COORDINATE}" \\
–conf spark.sql.crossJoin.enabled=true \\
–conf spark.sql.sources.default=parquet \\
–conf spark.sql.legacy.createHiveTableByDefault=true
4.3 Databricks
安装库（C++、Python、Scala 三层代码一次性装齐）：
集群 Spark 配置：进入集群 Configuration -> Advanced options -> Spark，在 Spark config 中加入：
spark.sql.crossJoin.enabled true
spark.sql.sources.default parquet
spark.sql.legacy.createHiveTableByDefault true
也可在 notebook 单元格中用 Python 等价设置：
spark.conf.set("spark.sql.crossJoin.enabled", "true")
spark.conf.set("spark.sql.sources.default", "parquet")
spark.conf.set("spark.sql.legacy.createHiveTableByDefault", "true")
准备本地缓存文件系统：要使用 C++ 快速预测，必须把 cache_path 指向一个所有 worker 都能访问的共享存储挂载点。在 Databricks 上典型做法是挂载 Azure Blob 存储：
dbutils.fs.mount(
source = "wasbs://<container>@<storage-account>.blob.core.windows.net",
mount_point = "/mnt/<container>",
extra_configs = {
"fs.azure.account.key.<storage-account>.blob.core.windows.net":
"<access-key>"
}
)
其中 <storage-account>、<container>、<access-key> 替换为实际值（例如存储账户 sarplusstorage、容器 sarpluscache 及其访问密钥）。随后构造模型时传入 cache_path="dbfs:/mnt/<container>/cache"，其中 cache 是缓存名。可选地，用 logging.getLogger("py4j").setLevel(logging.ERROR) 关闭 py4j 的日志噪音。
4.4 Azure Synapse
安装库：从 PyPI 下载 pysarplus 的 WHL 文件、从 Maven 中央仓库下载 sarplus JAR 文件（Spark 3.2+ 用 sarplus-spark-3-2-plus 版本），然后在 Synapse Workspace 的 Manage -> Workspace packages 中上传为工作区包，再到 Manage -> Apache Spark pools 中选中目标池，通过 … -> Packages -> Workspace packages -> + Select from workspace packages 选择并应用。pysarplus 也可通过 requirements.txt 安装。
准备本地缓存文件系统：Synapse 上需将 Spark 池 driver 节点的共享存储挂载到本地路径，例如挂载默认/主存储账户的容器：
from notebookutils import mssparkutils
mssparkutils.fs.mount(
"abfss://<container>@<storage-account>.dfs.core.windows.net",
"/mnt/<container>",
{ "linkedService": "<storage-linked-service>"}
)
job_id = mssparkutils.env.getJobId()
注意：Synapse 的 cache_path 必须以 synfs:/ 开头并在路径中带上 job_id（如 cache_path=f"synfs:/{job_id}/mnt/<container>/cache"），因为每个作业的本地挂载路径包含作业 ID。
4.5 三种环境的 cache_path 写法对照
| 本地机器 | cache_path="cache" | 直接作为本地目录使用 |
| Databricks（DBFS） | cache_path="dbfs:/mnt/sarpluscache/cache" | 源码会将 dbfs: 前缀替换为 /dbfs（SARPlus.py） |
| Azure Synapse | cache_path=f"synfs:/{job_id}/mnt/sarpluscache/cache" | 源码会将 synfs: 前缀替换为 /synfs |
该路径转换逻辑位于 SARPlus.py：Databricks 场景 dbfs:/… 映射到本地 /dbfs/…，Synapse 场景 synfs:/… 映射到本地 /synfs/…，普通路径原样使用。
五、C++ 快速预测链路详解
启用快速预测的完整调用链（SARPlus.py）：
C++ 端（pysarplus.cpp）的核心实现要点：
- MemoryMapFile 通过 open + fstat + mmap(PROT_READ, MAP_SHARED) 将 .sar 文件映射为只读共享内存（pysarplus.cpp），并在析构时 munmap/close 释放；
- SARModel 构造函数直接从映射地址解析布局：文件头部是行数（int64），随后是行偏移表 _offsets 与 item_score 数组 _related（pysarplus.cpp），零解析成本；
- predict 先对用户评分按物品 ID 排序，再遍历用户已见物品的邻接列表，跳过已见物品，用最小堆维护 Top-K（push_if_better），join_prod_sum 通过 lower_bound 二分查找对两个有序序列做归并求内积（pysarplus.cpp）；
- 通过 PYBIND11_MODULE(pysarplus_cpp, m) 暴露 SARModelCpp 与 SARPrediction（含只读 id/score 属性）给 Python（pysarplus.cpp）。
仓库内的单元测试（tests/test_pyspark_sar.py）覆盖了 C++ 预测的正确性验证：例如 test_good 中用户对物品 [0, 1] 的评分为 [10, 20] 时，Top-3 预测为 (0, 5.0)、(1, 44.0)、(2, 64.0)；test_good_require_sort 则验证了输入物品乱序时（[1, 0]）结果与有序输入一致，即 C++ 端排序预处理保证了顺序无关性。remove_seen=True 的剔除逻辑也有专门用例验证。
六、开发、测试与版本兼容性提示
若要在本仓库内自行构建和验证 SARplus：
- Python 包构建与测试（详见 DEVELOPMENT.md）：先构建 sdist，再对 Python 3.6～3.10 依次用 cibuildwheel 构建 manylinux 轮子，最后 pytest ./tests 运行 UDF + C++ 后端的测试；测试用例需要先编译出 sarplus JAR（测试 fixture 会在 scala/target 下查找）并启用 spark.sql.crossJoin.enabled 等配置（test_pyspark_sar.py）；
- Scala 侧测试：设置 SPARK_VERSION/HADOOP_VERSION/SCALA_VERSION 后执行 sbt ++${SCALA_VERSION}! test；
- Spark 3.2+ 注意：务必使用 sarplus-spark-3-2-plus 包，否则会因 OutputWriter.path() 破坏性变更而运行失败；
- 版本与依赖：pysarplus 的 setup.py 声明 python_requires=">=3.6,<3.12"，因此 Python 3.12 及以上环境不在支持范围内，安装前请核对运行环境。
七、常见问题与最佳实践
综上，SARplus 通过"PySpark 负责可扩展计算、C++ 负责极致推理性能、内存映射负责跨进程共享"的分层设计，为 SAR 这一经典邻域算法提供了适合生产环境的 Spark 落地路径。本文所有参数与代码行为均可对照仓库源码逐一验证，建议结合 contrib/sarplus/python/README.md、SARPlus.py 与 DEVELOPMENT.md 进一步深入。
赞
【免费下载链接】recommenders
Best Practices on Recommendation Systems
项目地址：
https://gitcode.com/gh_mirrors/re/recommenders
点击查看 免费下载
相关推荐
OpenCore Legacy Patcher终极指南：4步让老旧Mac焕发新生
ESP-IDF GATT Client API 详解：基于 Bluedroid 的 BLE 客户端开发指南
创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考
