推荐系统大规模特征工程与 FEDB 的 Spark 基于 LLVM 优化

2020-07-13
本文字数：4739 字
阅读完需：约 16 分钟

今天给大家分享第四范式在推荐系统大规模特征工程与 Spark 基于 LLVM 优化方面的实践，主要包括以下四个主题。

大规模推荐系统特征工程介绍
SparkSQL 与 FESQL 架构设计
基于 LLVM 的 Spark 性能优化
推荐系统与 Spark 优化总结

大规模推荐系统特征工程介绍

推荐系统在新闻推荐、搜索引擎、广告投放以及最新很火的短视频 App 中都有非常广阔的应用，可以说绝大部分互联网企业和传统企业都可以通过推荐系统来提升业务价值。

我们对常见的推荐系统架构进行分层，离线层（Offline layer）主要负责处理存在 HDFS 的大规模数据进行预处理和特征抽取，然后使用主流的机器学习训练框架进行模型训练并且导出模型，模型可以提供给在线服务使用。流式层（Stream layer）我们也称近线层，是介于离线和在线的中间层，可使用流式计算框架如 Flink 进行近实时的特征计算和生成，结果保存在 NoSQL 或者关系型数据库中给在线服务使用。在线层（Online layer）就包括了与用户交互的 UI 以及在线服务，通过实时的方式提取流式特征和使用离线模型进行预估，实现推荐系统在线的召回和推荐功能，预估结构和用户反馈也可以通过事件分发器写会流失计算的队列以及离线的 Hadoop 存储中。

本次分享重点会介绍离线层的优化，在大规模的推荐系统中离线存储的数据可能到 PB 级，常用的数据处理有 ETL（Extract, Transform, Load）和 FE（Feature extraction ），而编程工具主要是 SQL 和 Python，为了能够处理大规模数据一般使用 Hadoop、Spark、Flink 这样的分布式计算框架，其中 Spark 因为同时支持 SQL 和 Python 接口在业界使用最广。

SparkSQL 与 FESQL 架构设计

Spark 刚发布了 3.0 版本，无论是性能还是易用性都有很大的提升，其中相比于 Hadoop MapReduce 就有一百倍以上的性能加速，能够处理 PB 级数据量，支持水平拓展的分布式计算和自动 Failover，支持易用的 SQL、Python、R 以及流式编程（SparkStreaming）、机器学习（MLlib）和图计算（GraphX）接口，对于推荐系统来说内置的推荐算法模型也可以做到开箱即用。

业界使用 Spark 作为推荐系统离线数据处理框架的场景非常多，例如使用 Spark 来加载分布式数据集，使用 Spark UDF 和 SQL 来做做数据预处理和特征选择，使用 MLlib 来训练召回、排序模型。但是，在上线部分 Spark 就支持不了了。主要原因是 Spark 没有对 long running service 的支持，而 driver-executor 的架构只适合做离线的批处理计算，在 Spark 3.0 中推出了 Hydrogen 可以支持一些预先运行的 task 但也只适用于离线计算或者模型计算阶段，对于实时性要求更高但在线服务支持不好。Spark RDD 编程接口也是适合于迭代计算，我们总结下 Spark 的优势主要是能批量处理大规模数据，而且支持标准的 SQL 语法，劣势是没有在线预估服务的支持，因此也不能保证离线和在线服务的一致性，对于 AI 场景的特征计算也没有特别多的优化。

第四范式自研的 FESQL 服务，是在 SparkSQL 的基础上，提供了针对 AI 场景特征抽取计算的性能优化，还从根本上解决了离线在线一致性的问题。传统的 AI 落地场景是先在离线环境通过机器学习训练框架建模导出 AI 模型文件，然后由业务开发者来搭建在线服务，由于离线使用了 SQL、Python 进行了数据预处理和特征抽取等功能，在线需要开发一套与之匹配的在线处理框架，两套不同的计算系统在功能上容易出现离线在线不一致的情况，甚至离线建模时就可能使用穿越特征导致在线部分无法实现。而 FESQL 的方案是使用统一的 SQL 语言，除了标准的 SQL 支持外还拓展了针对 AI 场景的计算语法以及 UDF 定义，离线和在线使用同一套高性能 LLVM JIT 代码生成，保证了无论是离线还是在线都执行相同的计算逻辑，从而保证机器学习中离线和在线的特征一致性。

为了支持 SparkSQL 中无法支持的在线功能，FESQL 在线部分实现一个自研的高性能全内存时序数据库，相比于其他通用的 key-value 内存数据库如 Redis、VoltDB，在时序特征的存储上读写性能以及压缩容量都有很大的提升，并且比传统的时序数据库如 OpenTSDB 能够更好地满足在线服务超低延时的需求。而离线部分仍然借助 Spark 的分布式任务调度功能，只是在 SQL 解析和执行上使用了更高效的 native 执行引擎，通过 C++实现的 LLVM JIT 代码生成技术，可以针对 morden CPU 使用更多 intrinsic function 实现向量化等指令集优化，甚至是特有硬件如 FPGA、GPU 等加速。通过同一套 SQL 执行引擎等优化，不仅提升了离线和在线的执行效率，还能从功能上保证离线建模的特征抽取方案迁移到在线服务而不需要额外的开发和比对工作。

FESQL 性能上对比同样是全内存的商业产品 memsql，在针对机器学习的时序特征抽取场景中，同一个 SQL 在性能上相比 memsql 也有巨大的提升。

基于 LLVM 的 Spark 性能优化

从 Spark 2.0 开始，开始使用了 Catalyst 和 Tungsten 项目对 Spark 以及 SQL 任务有了很大的性能优化。Catalyst 通过对 SQL 语法进行词法解析、语法解析，生成了 unresolved 的抽象语法树数据结构，并且对抽象语法树进行了数十次的优化 pass，生成的最终物理计划可以比普通 SQL 解析后直接执行快数十倍。Tungsten 项目则是通过用 Java unsafe 接口实现了内部数据结构的堆外管理，在很大程度上降低了 JVM GC 的 overhead，并且对多个物理节点、多个表达式可以实现 whole stage codegen，直接生成 Java bytecode 并使用 Janino 内存编译器进行编译优化，生成的代码避免过多虚函数调用、提高 CPU cache 命中率，性能上比传统的火山模型解释执行快几倍并且非常接近由高级程序员手写 Java 代码的性能了。

那么 Spark 的 Catalyst 和 Tungsten 是否已经足够完美呢？我们认为还不够，首先 Spak 是基于 Scala、Java 实现的，就是是 PySpark 也是通过 socket 与 JVM 相连调用 Java 函数，因此所有代码都是在 JVM 上执行，这样不可避免就要接受 JVM 和 GC 的 overhead，而且随着 CPU 硬件和指令集的更新要通过 JVM 来使用新硬件特性还是比较困难的，更不用说越来越流行的 FPGA 和 GPU，对于高性能的执行引擎使用更底层的 C 或 C++实现可以代码更好的性能提升。对于可并行的数据计算任务，使用循环展开等优化手段可以成倍地提升性能，对于连续的内存数据结构还可以做更多向量化优化以及利用上 GPU 数千个计算核并行优化，这些在目前最新的 Spark 3.0 开源版中仍不支持。而且在机器学习场景中常用 SQL 的 window 函数来计算时序特征，这部分功能对应 Spark 的物理节点 WindowExec 居然没有实现 whole stage codegen，也就是说在做多表达式的划窗计算时无法使用 Tungsten 的优化，通过解释执行来计算每个特征，这样性能甚至比用户自己写的 Java 程序代码慢上不少。

为了解决 Spark 的性能问题，我们基于 LLVM 实现了 Spark 的 native 执行引擎，同时兼容了 SparkSQL 接口，相比与 Spark 会生成逻辑节点、生成 Jave bytecode，以及基于 JVM 运行在物理机上，FESQL 执行引擎也会解析 SQL 生成逻辑计划，然后通过 JIT 技术直接生成平台相关的机器码来执行，从架构上比 Spark 少了 JVM 虚拟机层的开销，性能也会有更大的提升。

LLVM 是目前非常流行的编译系统工具链，其中项目就包括了非常著名的 Clang、LLDB 等，而机器学习领域 TensorFlow 主推的 MLIR 以及 TVM 都使用了 LLVM 的技术，它可以理解为生成编译器的工具，目前很流行的 Ada、C、C++、D、Delphi、Fortran、Haskell、Julia、Objective-C、Rust、Swift 等编程语言都提供了基于 LLVM 实现的编译器。

JIT 则是与 AOT 的概念相对应，AOT（Ahead-Of-Time）表示编译是在程序运行前执行，也就是说我们常编写的 C、Java 代码都是先编译成 binary 或者 bytecode 后运行的，这就属于 AOT compiling。JIT（Just-In-Time）则表示运行时进行编译优化，现在非常多的解释型语言如 Python、PHP 都有应用 JIT 技术，对于运行频率非常高的 hot code 使用 JIT 技术编译成平台优化的 native binary，这种动态生成和编译代码技术也称为 JIT compiling。

LLVM 提供了高质量、模块化的编译、链接工具链，可以很容易实现一个 AOT 编译器，也可以集成到 C++项目中实现自定义函数的 JIT，下面就是实现一个简单 add 函数的例子，相比于直接用 C 来编写函数实现，JIT 需要在代码中定义函数头、函数参数、返回值等数据结构，最终由 LLVM JIT 模块来生成平台相关的符号表和可执行文件格式。由于 LLVM 内置了海量的编译优化 pass，自己实现的 JIT 编译器并不会比 GCC 或者 Clang 差很多，JIT 可用于生成各种各样的 UDF（User-Defined functions）和 UDAF（User-Defined Aggregation Functions），而且 LLVM 支持多种 backend，除了常见的 x86、ARM 等体系架构还可以使用 PTX backend 生成运行在 GPU 的 CUDA 代码，LLVM 还提供底层的 intrinsic functions 接口让程序可以用上现代的 CPU 指令集，性能与手写 C 甚至是手写 assembly 相当。

在 2020 年 Spark + AI Summit 上，Databrick 不仅 release 了 Spark 3.0，还提到了内部的闭源项目 Photon，作为 Spark 的 native 执行引擎可以加速 SparkSQL 等执行效率。Photon 同样使用 C++实现，从 Databrick 的实验数据可以看出，C++实现的字符串处理等表达式可以比 Java 实现的性能高出数倍，而且还有更多 vectorized 指令集支持。整体设计方案与 FESQL 非常类似，但 Photon 作为闭源项目目前只能在 Databrick 商业平台上使用，目前还在实验阶段需要联系客服手动开启，由于也没有更多实现细节公布因此不能确定 Photon 是否基于 LLVM JIT 实现，暂时官方也没有介绍有 PTX 或者 CUDA 的支持。

在表达式优化方面，主流和 Limit、Where 合并、Constant folding 以及 Filter、Cast、Upper、Lower 简化都可以通过 optimizationpass 来优化，生成最简洁的表达式计算从而大大减少 CPU 执行指令数，相关的 SQL 优化也就不一一赘述了，但只有经过逻辑节点优化、表达式优化、指令集优化、代码生成后才可能达到近于顶级程序员手写代码的性能。

在机器学习常用的时序特征抽取测试场景中，同一个 SQL 语句和测试数据，基于相同版本的 Spark 调度引擎，使用 FESQL 的 native 执行引擎性在单 window 下性能提升接近 2 倍，在更复杂的多 window 场景由于 CPU 计算更加密集性能可提升接近 6 倍，多线程下结果类似。

从结果上看，使用 LLVM JIT 的性能提升非常明显，使用相同的代码和 SQL，不会修改任何一行代码只要替换 SPARK_HOME 下的执行引擎实现，就可以实现接近 6 倍甚至更大的性能提升。我们从生成的计算图以及火焰图找到性能提升的原因，首先在 Spark UI 上可以看到，在 SparkSQL 中 window 节点是没有实现 whole stage codegen 的，因此这部分是 Scala 代码的解释执行，而且 SparkSQL 的物理计划很长，每个节点间 unsafe row 的检查和生成都有一定的开销，而对比 FESQL 只有两个节点，读数据后直接执行 LLVM JIT 的 binary 代码，节点间的 overhead 减少很多。从火焰图分析，底层都是 Spark 调度器的 runTask 函数，SparkSQL 在进行滑窗计算聚合特征时采样数和耗时都比较长，而 FESQL 是 native 执行，基本的 min、max、sum、avg 在编译器优化后 CPU 执行时间更短，左侧虽然有 unsafe row 的编解码时间但占比不大，整体时间比 SparkSQL 都少了很多。

FESQL 是目前少有的比开源 Spark 3.0 性能还更快数倍的 native 执行引擎，可以支持标准 SQL 以及集成到 Spark 中，与 Photon 仅能在 Databrick 内部使用不同，我们未来会发布集成 LLVM JIT 优化的 LLVM-enabled Spark Distribution，不需要修改任何一行代码只要指定 SPARK_HOME 就可以得到极大的性能加速，也可以兼容目前已有的 Spark 应用，更多 FESQL 使用案例请关注 Github 项目https://github.com/4paradigm/SparkSQLWithFeDB 。