Orange Financial 如何使用 Apache Pulsar 在每天超过 5000 万笔交易中打击金融欺诈
移动支付在中国取得了巨大的成功。如今，只需扫描二维码，即可在几秒钟内完成交易。这无疑给我们的日常生活带来便利的同时，移动支付也给风控基础设施带来了巨大挑战。 2019 年 9 月，我在纽约 O'Reilly Strata 数据大会上做了一个演讲，分享了我们公司如何利用 Apache Pulsar 来提高 Orange Financial 内部风险指标开发的效率。
关于 Orange 金融
甜橙金融（又名中国电信百世支付有限公司）是中国电信的关联公司。橘子金融成立于2011年3月，很快拿到了中国人民银行颁发的《支付业务许可证》。 Orange Financial旗下的子公司包括Bestpay、Orange Wealth、Orange Insurance、Orange Credit、Orange Financial Cloud等。特别是 Bestpay 已成为中国第三大支付提供商，紧随支付宝和微信支付之后。 Orange Financial 拥有 5 亿注册用户和 4190 万活跃用户，2018 年的交易量达到 1.13 万亿元人民币（183.7 亿美元）。
移动支付在中国
中国目前拥有全球最大的移动支付市场，而且还在逐年增长。 某研究院数据显示，2016年中国移动支付用户数为4.62亿，2019年达到7.33亿，2020年及以后还将继续增长。 2016 年移动支付交易总额为 22 万亿美元。到 2019 年底，预计将达到 45 万亿美元。
在中国，通过移动支付开展的经济活动数量激增，因为人们使用现金或信用卡的可能性比以往任何时候都小。 中国移动支付的高行业渗透率表明，移动支付与我们的日常生活息息相关。 你几乎可以用智能手机上的二维码做所有事情——订餐、乘坐出租车和地铁、租一辆自行车、买咖啡等等。
移动支付在中国取得了巨大的成功，因为它方便快捷。 只需扫描二维码，即可在数秒内完成交易。 这种速度和便利性加速了移动支付在电子商务、金融服务、运输、零售和其他业务中的采用。
我们的挑战
更易于使用会带来更大的威胁。移动支付在给我们的日常生活带来便利的同时，也给风控基础设施带来巨大挑战。即时交易涉及针对交易运行的数千条规则，以防止潜在的财务欺诈。 RSA 报告称，最常见的欺诈类型包括网络钓鱼、流氓应用程序、特洛伊木马攻击和品牌滥用。移动支付时代的金融威胁远不止于此。它们包括帐户或身份盗窃、商家欺诈和洗钱，仅举几例。根据中国银联的一项调查，在 105,000 名受访者中，有 60% 的受访者表示他们遇到过移动支付安全威胁。
我们拥有强大的风险管理系统，可帮助我们检测和预防这些攻击。然而，尽管近年来我们在保护客户资产方面做得很好，但我们仍然面临许多挑战：
- 高并发：我们的系统每天处理超过 5000 万笔交易和 10 亿个事件。峰值流量可以达到每秒 35,000 笔交易。
- 低延迟需求：我们要求我们的系统在 200 毫秒内响应交易。
- 大量的批处理作业和流式作业。
Lambda 架构
任何风险管理系统的核心都是决策。决策是一个或多个指标的组合，例如用户登录的地理坐标或零售商的交易量。例如，如果用户最近登录的地理坐标始终相同，则会引发怀疑。这告诉我们交易很可能是由机器人或模拟器发起的。同样，如果一个水果摊位的交易量在每天300美元左右，当交易量突然上升到3000美元时，我们的警报系统就会被触发。
制定风险控制指标既需要历史数据，也需要实时数据。比如我们总结一个商户过去一个月（30天）的总交易量，我们要以批处理的方式计算最近29天的交易量，然后和一个流式任务返回的值相加从当天上午 12 点开始收集的数据。
大多数互联网公司都部署了 Lambda 架构来解决类似的挑战。 Lambda 是有效的，并且在速度和可靠性之间保持了良好的平衡。之前，我们也采用了 Lambda 架构，它有三层：（1）批处理层，（2）流层，和（3）服务层。
批处理层用于历史数据计算，数据存储在 Hive 中。 Spark 是主要的批处理计算引擎。流层用于实时计算，Flink 是计算引擎，使用持久化在 Kafka 中的数据。服务层检索服务的最终结果。
Lambda 架构问题
然而，我们的经验表明，Lambda 架构存在问题，因为它复杂且难以维护。首先，我们必须将我们的业务逻辑拆分成很多部分。这增加了我们的通信开销并造成维护困难。其次，数据在两个不同的系统中重复，需要我们在不同的系统之间移动数据进行处理。
随着业务的增长，我们的数据处理栈变得非常复杂，因为我们必须不断地维护所有三个软件栈（批处理层、流层和服务层）。这也意味着我们必须维护多个集群：Kafka、Hive、Spark、Flink 和 HBase，以及拥有不同技能的多元化工程团队。这使得维护该数据处理堆栈的成本高得令人望而却步。
在寻求更有效的替代方案时，我们发现了 Apache Pulsar。借助 Apache Pulsar，我们大胆尝试重构我们的数据处理堆栈。目标是在我们的风险管理系统中简化堆栈、提高生产效率、降低成本并加速决策。
为什么 Apache Pulsar 效果很好
认识到简化业务流程和保持良好的财务风险控制所面临的独特挑战，我们开始研究 Apache Pulsar。
Apache Pulsar 是一个开源分布式发布-订阅消息系统，最初由 Yahoo! 创建。今天，它是 Apache 软件基金会的一部分。经过彻底调查，我们确定 Pulsar 最适合我们的业务。我们将得出这一结论的原因总结如下：
-
云原生架构和以段为中心的存储
Apache Pulsar 采用分层架构和基于段的存储（使用 Apache BookKeeper）。 Apache Pulsar 集群由两层组成： (a) 无状态服务层，由一组接收和传递消息的代理组成； (b) 有状态的持久层，由一组持久存储消息的 Apache BookKeeper 存储节点（称为“bookies”）组成。 Apache Pulsar 具有高可用性、强一致性和低延迟特性。
Pulsar 基于主题分区存储消息。每个主题分区都分配给系统中的一个活动代理（称为该主题分区的“所有者代理”）。所有者代理服务于从分区读取消息并将消息写入分区。如果代理失败，Pulsar 会自动将其拥有的主题分区移动到集群中剩余的可用代理。由于代理是“无状态的”，Pulsar 仅在节点故障或代理集群扩展期间将所有权从一个代理转移到另一个代理。重要的是，在此过程中不会发生数据复制。
-
2. Pulsar 主题分区上的消息存储在分布式日志中。该日志进一步分为多个段。每个段都存储为一个 Apache BookKeeper 分类帐，该分类帐分布并存储在集群内的多个 bookie 中。在以下三种情况中的一种情况下会创建一个新段：（1）在前一个段的写入时间超过配置的间隔（基于时间的滚动）之后； (2) 如果前一个段的大小已经达到配置的阈值（基于大小的滚动）；或 (3) 每当主题分区的所有权发生更改时。
通过分段，主题分区中的消息可以在集群中的所有 bookie 中均匀分布和平衡。这意味着主题分区的容量不仅仅受限于一个节点的容量。相反，它可以扩展到整个 BookKeeper 集群的总容量。
分层架构和以段为中心的存储（使用 Apache BookKeeper）是两个关键的设计理念。这些属性为 Apache Pulsar 提供了几个显着的好处，包括无限的主题分区存储、无需数据重新平衡的即时扩展以及服务和存储集群的独立可扩展性。
-
Apache Pulsar 提供两种读取 API：用于流式处理的 pub-sub 和用于批处理的段
Apache Pulsar 遵循一般的 pub-sub 模式。 (a) 生产者向主题发布消息； (b) 消费者订阅主题，处理收到的消息，处理完消息后发送确认（Ack）。
订阅是一个命名的配置规则，它决定了消息如何传递给消费者。 Pulsar 支持四种类型的订阅，它们可以在同一主题上共存，以订阅名称区分：
- Exclusive subscriptions：仅允许单个消费者附加到订阅。
- Shared subscriptions: 多个消费者可以订阅，每个消费者接收部分消息。
- Failover subscriptions: 多个消费者可以附加到同一个订阅，但只有一个消费者可以接收消息。只有当当前消费者失败时，排队的下一个消费者才开始接收消息。
- Key-shared subscriptions: 多个消费者可以附加到同一个订阅，具有相同密钥或相同排序密钥的消息仅传递给一个消费者。
在批处理过程中，Pulsar 采用以段为中心的存储，从存储层（BookKeeper 或分层存储）读取数据。
-
使用 Pulsar 和 Spark 构建统一的数据处理堆栈
了解了 Apache Pulsar 之后，我们选择了该产品来构建一个新的统一数据处理栈，使用 Pulsar 作为统一数据存储，Spark 作为统一计算引擎。
Spark 2.2.0 Structured Streaming 为批处理和流式处理提供了坚实的基础。 您可以通过 Spark Structured Streaming 读取 Pulsar 中的数据，并通过 Spark SQL 查询 Pulsar 中的历史数据。
Apache Pulsar 通过将数据存储在分段流中来解决其他系统的混乱操作问题。 数据在到达时附加到主题（流），然后分段并存储在可扩展的日志存储中，Apache BookKeeper。 由于数据仅存储为一个副本（“事实来源”），它解决了 Lambda 架构中的不一致问题。 同时，我们可以通过统一的发布-订阅消息和分段访问流中的数据，以进行弹性并行批处理。 与 Spark 这样的统一计算引擎一起，Apache Pulsar 是构建统一数据处理堆栈的完美统一消息和存储解决方案。 鉴于这一切，我们决定采用 Apache Pulsar 为我们的业务重新构建我们的堆栈。
迁移到 Apache Pulsar
要为我们的业务启用 Apache Pulsar，我们需要升级我们的数据处理堆栈。升级分两步完成。
首先，从旧的基于 Lambda 的数据处理堆栈导入数据。我们的数据由历史数据和实时流数据组成。对于实时流数据，我们利用 pulsar-io-kafka 从 Kafka 读取数据，然后写入 Pulsar，同时保持模式信息不变。对于历史数据，我们使用 pulsar-spark 查询 Spark 存储在 Hive 中的数据，并将结果以 Schema (AVRO) 格式存储到 Pulsar 中。 pulsar-io-kafka 和 pulsar-spark 都已经被 StreamNative 开源了。
其次，移动我们的计算作业来处理存储在 Pulsar 中的记录。我们使用 Spark Structured Streaming 进行实时处理，使用 Spark SQL 进行批处理和交互式查询。
新的基于 Apache Pulsar 的解决方案统一了计算引擎、数据存储和编程语言。与 Lambda 架构相比，新解决方案显着降低了复杂性：
- 将复杂度降低 33%（集群数量从六个减少到四个）；
- 节省存储空间8.7%（预期：28%）；
- 生产效率提升11倍（支持SQL）;
- 由于统一架构，稳定性更高。
结论
Apache Pulsar 是一个云原生消息传递系统，具有分层架构和以段为中心的存储。 Pulsar 是构建我们统一数据处理堆栈的完美选择。 与 Spark 这样的统一计算引擎一起，Apache Pulsar 能够提高我们风险控制决策部署的效率。 因此，我们能够为商家和消费者提供安全、便捷、高效的服务。
Pulsar 是一个年轻而有前途的项目，Apache Pulsar 社区正在快速发展。 我们在新的基于 Pulsar 的统一数据堆栈上投入了大量资金。 我们希望将我们的实践贡献给 Pulsar 社区，并帮助面临类似挑战的公司解决他们的问题。
参考
