Angel 分布式 Swing 算法实战：基于用户行为计算 Item 相似度的推荐召回方案-开发者社区

人工智能
机器学习
分布式训练
图计算
后端

【免费下载链接】angel

A Flexible and Powerful Parameter Server for large-scale machine learning

项目地址：https://gitcode.com/gh_mirrors/an/angel

本文以 Angel 开源项目（Spark on Angel）中的 Swing 图算法为例，系统讲解如何利用用户行为二部图计算 item 之间的相似度，并输出itemId itemId score形式的相似度结果用于推荐召回。读者将掌握 Swing 的相似度公式与各参数（alpha/beta/gamma/topFrom/topTo 等）的真实含义、Spark on Angel 集群任务提交方式、内存资源估算方法，以及常见超时问题的排查思路。文中涉及的公式、参数与执行流程均可在仓库源码中找到对应实现。

1. 算法介绍：为什么"共同购买者越少，Item 越相似"

Swing 是一种利用用户行为（如购买、点击）计算 item 与 item 之间相似度的算法，通常用于推荐系统的召回阶段。其基本原理是：如果两个 item 的购买用户集合（共同关联的 user）越少，那么这些 item 之间的相似度越高。也就是说，一对 item 被"少数几个口味一致的忠实用户"共同购买，比被"大量泛化用户"共同购买更能说明它们之间的相似关系——后者往往来自用户的普适偏好，区分度低。

在二部图中，记Ui为购买过 item_i 的用户集合，Uj为购买过 item_j 的用户集合，Iu为用户 u 购买的 item 集合，Iv为用户 v 购买的 item 集合，Swing 相似度计算公式如下：

公式包含两部分：

相似度累加项：sim(i,j) = Σ W / (alpha + |Iu ∩ Iv|)，对同时购买过 item_i 与 item_j 的用户对 (u, v) 求和。分母中的|Iu ∩ Iv|是用户 u 与用户 v 共同购买过的 item 数量，公共购买越多，单对用户的贡献越小；
权重项：W = (|Iu| + beta)^gamma × (|Iv| + beta)^gamma，其中gamma通常取值范围为[-1, 0]（英文文档记为[-1, 0)），beta与gamma共同构成对两条购买记录长度的惩罚——用户购买列表越长，其"口味信号"越弱，贡献权重越低。

上述公式在源码 SwingOperator.scala 中实现为：

def Score(Iu:Array[Long], Iv:Array[Long], alpha:Float, weight: Float = 1.0f): Float = { weight / (alpha + ArrayUtils.intersectCount(Iu, Iv)) }

而权重wj = (|Iu| + beta)^gamma × (|Iv| + beta)^gamma由swing()方法中的math.pow(uItems.length + beta, gamma)与math.pow(vItems.length + beta, gamma)逐对计算（见 SwingOperator.scala）。可以看到，alpha作为分母平滑项存在，beta对购买列表长度做偏移，gamma为负值时即对长购买记录降权。

2. 输入输出与数据格式

Swing 的输入是一张无权的 user-item 二部图，输出是 item 之间的相似度结果，两者均为 HDFS 上的文本文件：

参数	含义	说明
`input`	输入二部图（HDFS 路径），不带权	每行表示一条边，格式为`userId itemId`，由`sep`指定的分隔符分开
`output`	输出（HDFS 路径）	每行表示一对 item 及其相似度，格式为`itemId itemId score`
`sep`	分隔符	输入中每条边的起始顶点、目标顶点之间的分隔符，如`tab`、空格等

示例输入：

10001 20001 10001 20002 10002 20001 10002 20003

示例输出（score为相似度，itemI itemJ与itemJ itemI都会写出对称结果）：

20001 20002 0.125 20002 20001 0.125

在源码中，输入由 SwingExample.scala 通过GraphIO.load(input, isWeighted = false, srcIndex, dstIndex, sep)加载，输出通过GraphIO.save(mapping, output)保存。srcIndex/dstIndex可指定边文件中源点（user）与终点（item）所在的列索引，默认均为 0、1。

3. 算法参数详解：默认值、取值范围与调优含义

3.1 核心算法参数

参数	含义	默认值
`topFrom`	将 item 按出现次数（即被多少用户购买）由大到小排序后，仅计算排名在`[topFrom, topTo)`之间的 item 与其他所有 item 的相似度，适用于只计算尾部（长尾）商品之间的相似度	0
`topTo`	参见`topFrom`解释，构成排名区间上界	0
`alpha`	公式中的平滑项，对应源码中的 delta 参数	0
`beta`	公式中对购买记录长度的偏移惩罚参数	5
`gamma`	公式中对购买记录长度的幂次惩罚参数，取值范围`[-1, 0]`	-0.3
`partitionNum`	数据分区数，Spark RDD 数据的分区数量	1
`psPartitionNum`	参数服务器（PS）上模型的分区数量	取`spark.ps.instances`
`useBalancePartition`	参数服务器对输入数据节点存储划分是否采用均衡分区。如果输入节点的索引分布不均匀，建议选择 true	—
`storageLevel`	RDD 存储级别，可选`DISK_ONLY`/`MEMORY_ONLY`/`MEMORY_AND_DISK`	`MEMORY_ONLY`

参数默认值在仓库中有多处印证：

HasBeta.scala 中setDefault(beta, 5)；
HasGamma.scala 中setDefault(gamma, -0.3f)；
示例入口 SwingExample.scala 中alpha默认"0"、beta默认"5"、gamma默认"-0.3"、storageLevel默认MEMORY_ONLY。

调优建议（由公式与文档推断）：

gamma越接近 -1，对购买列表长的用户惩罚越重，相似度越倾向于由"短列表、高忠诚度"用户贡献；
topFrom/topTo组合适合关注长尾 item 的相似度（例如新上架商品召回），当区间覆盖 item 数超过 1 亿时，源码 Swing.scala 会打印提示信息；
storageLevel用于控制 user-item 邻接表 RDD 的持久化级别，内存紧张时可切到DISK_ONLY换取稳定性。

3.2 示例独有的进阶参数

集群示例 SwingExample.scala 还解析了以下进阶参数（文档未逐一列出，但同样可通过命令行传入）：

参数	含义	示例默认值
`batchSize`	向 PS 初始化邻接表时每个批次处理的节点数	10000
`pullBatchSize`	计算阶段从 PS 批量拉取邻居的分批大小	200
`superItemThreshold`	共同用户数超过该阈值的 item 对会进入"超级 item 对"单独批处理通道	2000
`superItemPairBatch`	超级 item 对的批处理大小	800
`srcIndex`/`dstIndex`	输入文件中源点、终点的列索引	0 / 1
`cpDir`	Spark checkpoint 目录	默认取`GraphIO.defaultCheckpointDir`

其中superItemThreshold与superItemPairBatch对应 Swing.scala 中的"超级 item 对"处理逻辑：当某 item 对的共同用户数超过阈值时，直接按(item_i, item, -1f)标记，再统一交给calcSuperItemPairs以批处理方式补算分数，避免单个分区内出现超大规模集合运算。

4. 资源估算与内存配置

Swing 在 PS 上存储的是user-item 邻接表（每个用户节点挂其购买过的 item 列表），内存占用可以按边数估算。文档给出了明确的内存配比建议：

PS 侧：ps.instance与ps.memory的乘积是 PS 总的配置内存。为保证 Angel 不因内存不足而挂掉，需要配置约为 PS 上数据存储量2 倍左右的内存；
Spark Executor 侧：num-executors与executor-memory的乘积是 executors 总的配置内存，最好能存下2 倍的输入数据。如果内存紧张，1 倍也可以接受，但运行会相对慢一些。

文档给出的估算示例：100 亿条边集的输入大约有 160G 大小，20G × 20（即 20 个 executor、每个 10G 内存左右）的配置是足够的；在资源实在紧张的情况下，尝试加大分区数目（partitionNum）来分摊压力。

这一内存模型与源码的执行流程一致：edges先被persist(StorageLevel.DISK_ONLY)落盘（见 Swing.scala），随后userItemNeighborTable按storageLevel持久化并 push 到 PS，计算阶段每个 partition 再从 PS 按pullBatchSize批量拉取邻居——PS 上的邻接表大小直接决定 ps.memory 需求。

5. 任务提交示例（Spark on Angel）

5.1 集群模式提交

以下为文档提供的完整 Yarn 集群提交示例（参数通过key:value形式跟在 jar 之后传入）：

input=hdfs://my-hdfs/data output=hdfs://my-hdfs/output source ./spark-on-angel-env.sh $SPARK_HOME/bin/spark-submit \ --master yarn-cluster \ --conf spark.ps.instances=1 \ --conf spark.ps.cores=1 \ --conf spark.ps.jars=$SONA_ANGEL_JARS \ --conf spark.ps.memory=10g \ --name "swing angel" \ --jars $SONA_SPARK_JARS \ --driver-memory 5g \ --num-executors 1 \ --executor-cores 4 \ --executor-memory 10g \ --class org.apache.spark.angel.examples.graph.SwingExample \ ../lib/spark-on-angel-examples-3.3.0.jar input:$input output:$output sep:tab storageLevel:MEMORY_ONLY useBalancePartition:true \ partitionNum:4 psPartitionNum:1

其中spark.ps.*系列配置用于启动 Angel PS（参数服务器）：spark.ps.instances=1表示 1 个 PS 实例、spark.ps.cores=1分配 1 个核、spark.ps.memory=10g分配 10G 内存；$SONA_ANGEL_JARS/$SONA_SPARK_JARS由spark-on-angel-env.sh环境脚本注入，对应 Angel 与 Spark 的依赖 jar。当前仓库中与文档同名入口对应的示例源码位于 cluster/SwingExample.scala，包名为com.tencent.angel.spark.examples.cluster.SwingExample，任务提交的通用环境准备可参考 spark-on-angel/README.md 与 spark_on_angel_quick_start.md。

5.2 本地模式运行

仓库还提供了本地模式示例 local/SwingExample.scala，通过mode参数（默认yarn-cluster）切换运行方式，便于在小数据上快速验证：

# mode 可指定为 local 等本地 master --mode local --input <本地或HDFS输入> --output <输出路径> sep:tab storageLevel:MEMORY_ONLY partitionNum:4 psPartitionNum:1

6. 分布式执行流程：从二部图到相似度结果

结合 Swing.scala 的transform方法，一次 Swing 任务在分布式环境下的完整流程如下：

读取边数据：NeighborDataOps.loadEdges读取 user-item 边并落盘DISK_ONLY，打印抽样结果；
（可选）筛选目标 item 区间：当topTo > topFrom时，对 item 按出现次数降序排序并取排名[topFrom, topTo)的 item 集合（见 Swing.scala）；
构建并统计邻接表：SwingOperator.userItem2NeighborTable将边按 key 分组为(节点, 邻居数组)，其中邻居数组做distinct.sorted去重排序（见 SwingOperator.scala）；stats方法统计出 min/max id、节点数、边数与最大/最小度数；
初始化 PS 模型：PSContext.getOrCreate启动 PS，创建SimpleNeighborTableModel，按batchSize分批将 user-item 邻接表 push 到 PS 并checkpoint()；
分区并行计算相似度：itemUserNeighborTable.mapPartitionsWithIndex在每个分区内，按pullBatchSize从 PS 批量拉取用户邻居，再调用swing()计算 item 对分数（见 SwingOperator.scala）；
超级 item 对补算：共同用户数超过superItemThreshold的 item 对进入calcSuperItemPairs批处理通道，聚合后与主结果union；
输出：过滤掉标记为-1f的中间结果，按(itemI, itemJ, score)三元组写出 DataFrame（见 Swing.scala）。

值得注意的实现细节：swing()对每对用户计算权重wj时使用了math.pow(uItems.length + beta, gamma)（Scala 的math.pow返回 Double），并与交集计数结合得到wj / (alpha + intersectCount(Iu, Iv))；对称的 item 对(item, item_i)也会被写出，保证输出结果对每对 item 双向可见，便于直接用于召回索引。

7. 常见问题排查

文档记录了一个典型的高频故障：任务运行约 10 分钟时挂掉。

可能原因：Angel 申请不到资源。由于该任务基于 Spark on Angel 开发，实际涉及 Spark 和 Angel 两个系统，向 Yarn 申请资源时是独立进行的——Spark 任务拉起之后，由 Spark 向 Yarn 提交 Angel 的任务，如果不能在给定时间内申请到资源，就会报超时错误导致任务挂掉；
解决方案：
1. 确认资源池有足够的资源；
2. 添加 Spark confspark.hadoop.angel.am.appstate.timeout.ms=xxx调大超时时间，默认值为600000，也就是 10 分钟。

此外，结合第 4 节的资源估算，若任务频繁 OOM 或执行缓慢，应优先检查 PS 内存（ps.instance × ps.memory）是否达到邻接表存储量的 2 倍、executor 总内存是否能容纳 2 倍输入数据，并在内存紧张时增大partitionNum分区数。

小结

Swing 算法以"共同购买者越少越相似"为核心思想，通过 alpha/beta/gamma 三个参数实现对购买记录长度的平滑与惩罚，特别适合长尾 item 的相似度召回。在 Angel 的 Spark on Angel 框架下，Swing 将邻接表存放在 PS 上、以分区并行方式批量拉取计算，兼顾了超大二部图的可扩展性。结合本文给出的参数默认值、内存估算公式与提交脚本，读者可以直接在 Yarn 集群上跑通input -> output的完整链路。更多细节可进一步阅读仓库中的 swing_en.md 英文文档以及 Swing.scala 源码实现。