基于 Spark 的大规模推荐系统特征工程

导读： 特征工程在推荐系统中有着举足轻重的作用，大规模特征工程处理的效率极大的影响了推荐系统线上的性能。第四范式作为国际领先的机器学习和人工智能技术与平台服务提供商，面向大规模特征工程问题开发了下一代离线在线一致性特征抽取引擎 FESQL，针对 AI 场景支持 SQL 接口，兼容 Spark 3.0 同时提供高性能的 Native 执行引擎。本次分享题目为基于 Spark 的大规模推荐系统特征工程及优化，主要内容包括：

• 大规模推荐系统

• Spark SQL 应用与 FESQL

• 基于 LLVM 的 Spark 优化

• 总结

01 大规模推荐系统

1. 业界推荐系统的应用

众所周知，推荐系统在业界有着许多成功的应用，据统计，亚马逊 40%的销售在推荐系统的作用下产生；Netflix 75%的用户使用推荐系统寻找他们喜爱的视频；30%的用户进行在线购物前会使用关键词搜索他们需要的商品。目前，几乎所有的新闻、搜索、广告、短视频应用都是基于推荐系统建立的。

2. 推荐系统的架构

业界成熟的推荐系统架构一般分为三层：离线层 ( offline layer )，近实时的流式层 ( stream layer ) 和在线层 ( online layer ) 三部分。

离线层： 一般用于大规模的数据预处理、特征抽取与模型训练，通常用 Hadoop HDFS 进行数据存储，使用 Spark，MapReduce 等分布式计算引擎进行特征抽取与计算以及数据管理，再使用离线模型训练框架 TensorFlow、Pytorch、MXNet 等进行离线的模型训练，模型结果可用于线上预测。

近实时的流式层： 主要是为了提升推荐系统的时效性，对于一些时序特征，可以使用消息队列收集近实时的数据，结合流式计算服务如 Flink 对数据进行补全，把结果存入 NoSQL、MySQL 等存储服务中，存储结果供线上服务使用。

在线层： 用户产生的数据可以通过 Flink 生成流式特征，也可以使用 HDFS 进行数据归档。在线预估时从 NoSQL 或 MySQL 中提取流式特征，通过离线训练的模型即可进行线上预估。

3. 大规模推荐系统的特征抽取

大规模推荐系统的数据处理通常分为两类：

• ETL ( Extract, Transform, Load )：进行数据数据补全、格式转换等；

• 特征抽取：对原始数据特征进行处理，得到模型易于学习的样本特征，如离散化，embedding 化等方法。

常用工具包括：

• SQL/Python：针对一般规模的数据，通常可以通过使用 SQL/Python 进行处理；

• Hadoop/Spark/Flink：针对大规模数据，通常要借助 Hadoop/Spark/Flink 等计算框架。

02 Spark SQL 应用与第四范式自研 FESQL 技术

1. Spark 简介

Spark 是专为大规模数据处理而设计的快速通用的计算引擎，依托强大的分布式计算能力，在 Spark 上可以开发机器学习、流式学习等应用。Spark 提供了 SparkSQL，使其能与 SQL、Hive 兼容，提供 PySpark 接口可以让开发者使用 Python 进行分布式应用开发，提供了 MLlib 包，可以用于机器学习应用的开发。同时 Spark 也提供诸如 Catalyst/Tungsten 等方式的优化。

Spark 的优势就在于：计算速度快，能够处理 PB 级别的数据，分布式计算和自动容错机制，提供便于使用的 SQL/Python/R API，同时，Spark 提供的机器学习库也可以应用于推荐系统，所以在业界，几乎所有公司都会使用 Spark 作为离线层数据处理框架。

2. 大规模推荐系统中的 Spark 应用

以 IBM 的一个推荐系统开源项目来说明 Spark 在推荐系统中的应用。首先是数据加载，使用 read.csv 即可加载本地或 HDFS 数据。使用 select 即可进行特征列选择。

然后是对数据进行预处理以及简单的特征抽取，该项目中使用了 Spark UDF 对字符串进行处理，抽取出其中的年份信息，将年份信息作为特征进行使用。

得到全部特征预处理的结果后即可进行模型训练，可以使用 Spark 内置机器学习 API 进行模型训练。训练完成后，模型即可上线进行线上预估。

线上的预估服务需要提供实时计算的预估接口，但是在实践中，Spark 并不适合直接用于线上预估。原因有三：

• Driver-exexutor 结构只适合进行批量处理，不适合在线处理

• Spark 的批处理模式不适合提供长时间运行的在线服务，也不能保证低延时的计算效率（Spark 3.0 的 Hydrogen 可以部分支持）

• RDD 接口只适合迭代计算，不适合做实时计算

因此，业界的通常做法是使用 Java、C++等后端语言实现在线的预估服务，这就带来了另一个线上特征抽取的一致性问题，由于必须要保证线上线下特征的一致性，所以必须同时开发线上使用的特征处理模块，并人工保证计算结果没有差异。

3. Spark 的优缺点

Spark 支持大规模数据的批处理，提供标准的 SQL 接口的优点使其成为离线层数据处理的不二之选，但是，Spark 不支持线上服务，不能保证线上线下特征一致性，同时在 AI 场景下的性能没有经过优化，所以在 AI 场景下，Spark 仍有许多不足。针对这些不足，第四范式开发了 FESQL 执行引擎。

4. FESQL 线上线下一致性执行引擎

FESQL——保证离线在线特征一致性的 SQL 执行引擎。上图表示传统的上线过程，生成离线模型文件后，由应用开发者开发线上预估服务，将 Spark、SQL 中的特征处理逻辑翻译成后端语言代码，实现线上服务，每新增一个特征，都要开发对应的特征抽取模块，同时需要用户和业务开发者保证特征数据的一致性。下图是使用 FESQL 的上线过程，由于线上线下使用统一的 SQL 服务进行特征抽取，因而保证了特征在线上和线下的一致性。

图中所示为 FESQL 基本框架，左边离线部分和 SparkSQL 的用法基本一致，由数据科学家设计 SQL 语句，基于 Spark 进行离线批处理。橙色框表示第四范式开发的基于 LLVM 优化的 SQL 引擎，性能大大优于原生 Spark，同时能够更好的支持线上服务，尤其对于 SQL 语句进行了拓展，使之能够更好的支持机器学习场景下的线上特征处理。其中 FEDB 是有第四范式开发的全内存数据库，相比于 Spark 读取 HDFS 这种高延时的数据载入方式，FEDB 可以提前载入模型预估所需数据，效果接近开发的线上特征抽取模块，同时支持时序特征。线上线下的数据一致性由同一套的 SQL 执行引擎保证。

5. 性能对比

与兼容 SQL 的全内存数据库 memsql 的方式进行性能对比可以发现，LLVM 优化后的 SQL 之心引擎在读和写的性能上都要更高。

对于机器学习场景下的列聚合 ( 生成时序特征 ) 场景，LLVM 优化后的 SQL 引擎也比 memsql 快很多，耗时基本小于 memsql 的 50%。

03 基于 LLVM 的 Spark 优化

1. Spark Catalyst 和 Tungsten 优化

Spark2.0 之后提供了 Catalyst 和 Tungsten 优化。图为 Catalyst 从 SQL 解析到生成物理计划的流程图，由 SQL 语句或 DataFrame 接口通过编译器技术 ( 语法解析等 ) 生成 Unresolved Logical Plan，Catalyst 通过解析 Catalog 对 Unresolved Logical Plan 处理得到 Logical Plan，在经过 SQL 常用优化方案，得到 Optimized Logical Plan，优化之 Catalyst 后可以生成多个基于 Spark 运行的 Physical Plan，最终选择其中最高效的进行运行。该方式适合于计算节点优化，对于 SQL 的优化也同样效果显著。

Tungsten 是另外一种优化方案。主要的优化点在于：

• 内存管理与堆外存储避免了多余的内存使用，同时减少了 GC；

• 引入 code generation 技术，通过 JIT 编译运行，Spark 动态生成 Java 字节码来计算这些表达式，而不是为逐行解析执行，减少了原始数据类型的装箱操作，更重要的是避免了 Overhead 较大的虚函数调用。

以一个经典实例来介绍 Tungsten 的原理。左侧的 SQL 命令可以翻译成在 Spark 上运行的 Logical Plan，由下往上分为 4 个计算节点，传统的 SQL 执行引擎中，四个节点分别由四个迭代器实现 ( 可以理解为四个循环 )，循环没有合并优化以及节点的虚函数调用对于 CPU Cache 非常不优化，导致传统的 SQL 引擎计算性能比较差。右侧为 Tungsten 优化后的结果，使用了 whole staged code generation，对多节点的循环进行了合并，性能有着明显的提升。

2. Catalyst/Tungsten 的不足

Catalyst/Tungsten 给 Spark 带来了明显的性能能提升，但 Catalyst/Tungsten 的优化仍然是基于 Java 进行的，如果能使用更底层的指令集，如汇编、二进制码效果会更好；JVM 难以支持循环展开等优化方式；而且并非所有的节点都支持 code generation，例如图中的 WindowExec 节点就不支持 code generation。

3. FESQL

鉴于以原因，Catalyst/Tungsten 的优化仍有不足，第四范式基于 LLVM 技术进一步优化得到 FESQL。SparkSQL 架构如黄色部分所示，FESQL 架构如蓝色框所示，根据 SparkSQL 语句生成 FESQL Logical Plan，再由 LLVM JIT 生成平台二进制码直接执行，相比于 Spark 少了 JVM 一层，性能也会有明显提升。

4. LLVM 简介

LLVM 项目是一个模块化的、可重用的编译器和工具链集合，可以方便的实现编译器和代码生成的工作。提供了许多有用的工具，如 Clang、LLDB、MLIR、TVM 等，能够实现多种编程语言的编译器。

JIT ( Just-In-Time Compiler ) 编译，可以一边运行程序一边编译二进制代码，右图为使用 JIT 编译的 Add 函数，这部分代码可以在运行时被翻译成底层代码，与直接使用 C++来实现效率接近，同时 JIT 能够适应不同的 CPU 生成优化的二进制码。

5. FESQL 的优化点

目前已经能使用循环展开、常数折叠、向量化和一些基于 CPU 本身的优化；未来，基于 PTX 后端还可以尝试生成 CUDA 代码，利用 GPU 进行计算的加速。

6. 性能比较

FESQL 与 Databrick 内部的 Photon 非常相似 ( Photon 内部由 C++实现 )，因而进行对两者进行比较。Photon 是 Databrick 的企业产品，仅能在 Databrick 的平台上使用，且不支持 PTX/CUDA。对比由 C++和由 JVM 实现的处理引擎的性能，发现 C++实现的处理引擎性能非常优越。

7. FESQL 的节点优化

FESQL 使用了节点优化，使用 SimpleProject 对 Project 节点进行合并优化，对窗口节点使用 code generate 进行优化。下图说明了对于节点的优化可以明显减少执行的流程。

8. FESQL 的表达式优化

FESQL 也实现了非常多表达式优化，保证在不同 SQL 场景都比传统数据库有着更好的性能表现。

9. 性能

对比 Spark 3.0 和 FESQL on Spark 可以发现，FESQL 的执行效率明显高于 Spark 3.0，多窗口的情况下效果更明显，有着接近 6 倍的性能提升。

通过对比两者生成的逻辑计划图，可以发现 FESQL 的计划图明显更简单，通过对比两者的火焰图，底层 RDD 计算基本一致，FESQL 取样的样本数更少，执行时间更短，因此 FESQL 的执行效率更高。

10. 展望

未来第四范式计划推出 LLVM-enabled Spark Distribution，使开发者可以通过设置 SPARK_HOME 便利的实现性能加速；为开发者提供 Docker、Notebook、Jar、Whl 包，便于开发；提供类似 Python 的保证一致性的 DSL 语言用于 UDF 和 UDFA 实现；还有提供对 CUDA 和 GPU 的支持。

04 总结

大规模推荐系统中可以使用 Spark、Flink、ES、FESQL 实现大规模的数据处理，其中 Spark 更适合离线的批处理，而不适合线上处理，FESQL 能同时进行线上线下服务因为能够保证特征一致性，同时 LLVM JIT 实现的 FESQL 拥有比 Spark 3.0 更好的性能。

作者介绍：

陈迪豪，第四范式架构师

第四范式先知平台架构师，负责深度学习框架产品化以及下一代特征引擎开发工作。积极参与了开源社区 TensorFlow、Kubernetes、TVM 等项目开发，对分布式系统和深度学习平台有一定了解，目前专注于离线在线一致性的特征引擎开发。

本文来自 DataFunTalk

原文链接：

基于Spark的大规模推荐系统特征工程