使用 Apache Cassandra 和 Apache Pulsar 在零售业实施人工智能

实施人工智能和机器学习解决方案的过程需要解决数字系统中经常出现的许多常见挑战：更新遗留系统、消除批处理流程以及使用基于人工智能/机器学习的创新技术来改善客户体验。就在几年前，这似乎还像科幻小说中的那样。

为了说明这一演变，让我们假设一个承包商受雇帮助一家大型零售商实施 AI/ML 解决方案。这是系列文章中的第一篇，将详细介绍 AI/ML 之旅的重要方面。

问题：脆弱的批处理过程

这是 BigBoxCo 的“基础设施”团队的第一天。在完成了强制性的人力资源活动后，我收到了承包商徽章并前往我的新工作空间。

与团队会面后，我被告知今天早上我们将与“建议”团队举行会议。

我的系统访问权限尚未完全正常，因此希望 IT 部门能够在我们开会时解决这个问题。

会议室里只有我们几个人：我的经理和新团队的另外两名工程师，以及推荐团队的一名工程师。我们首先进行一些介绍，然后继续讨论前一周的问题。

显然，上周发生了某种隔夜批量失败的情况，他们仍然感受到了其影响。

目前的产品推荐似乎是由从客户订单收集的数据驱动的。对于每个订单，所订购的产品之间都会有一个新的关联，并进行记录。

当客户查看产品页面时，他们可以根据有多少其他客户购买了当前产品以及不同的产品来获得推荐。

产品推荐通过云中的微服务层提供给 bigboxco.com 上的用户。微服务层使用Apache Cassandra的本地（云）数据中心部署来提供结果。

然而，如何收集和提供结果则完全是另一回事。本质上，产品（一起购买）之间的关联结果是在 MapReduce 作业期间编译的。这是上周失败的批处理过程。

虽然这个批处理过程从来都不是很快，但随着时间的推移，它变得更慢、更脆弱。事实上，有时该过程需要两天甚至三天的时间才能运行。

改善体验

会议结束后，我检查了电脑，看来我终于可以登录了。正当我环顾四周时，我们的首席工程师（PE）过来做了自我介绍。

我告诉他与推荐团队的会议，他向我介绍了更多推荐服务背后的历史。

听起来这个批处理过程已经存在了大约 10 年。设计它的工程师已经离开了，组织中没有多少人真正理解它，也没有人愿意碰它。

我开始解释的另一个问题是，驱动每个建议的数据集几乎总是几天前的。

虽然从总体上看这可能不是什么大问题，但如果推荐数据能够更新，将有利于营销的短期促销。

他点头同意，并表示他绝对愿意接受有关如何改进系统的建议。

也许这是一个图表问题？

一开始，这对我来说听起来像是一个图形问题。我们有登录网站并购买产品的客户。在此之前，当他们查看产品或将其添加到购物车时，我们可以以“购买 X 的客户也购买了 Y”的形式显示推荐。

该网站今天已经做到了这一点，因为推荐服务正是这样做的：它返回经常一起购买的前四种附加产品。

但我们必须有某种方法对产品进行“排名”，因为我们 2 亿客户中的任何一个客户同时购买的一种产品与其他所有其他产品的映射将会变得越来越大、越来越快。

因此，我们可以根据它们在订单中出现的次数对它们进行排名。该系统的图表可能类似于图 1 所示。

在对其进行建模并使用实际数据量在我们的图形数据库上运行之后，我很快意识到这是行不通的。

从一种产品到附近客户的产品的遍历以及计算出现次数最多的产品大约需要 10 秒。

本质上，我们“押注”了两天的批量问题，让每次查找都将遍历延迟精确地放在我们不想要的地方：在客户面前。

但也许这个图模型与我们在这里需要做的并不太远？事实上，上述方法是一种称为“协作过滤”的机器学习 (ML) 技术。

从本质上讲，协作过滤是一种根据与其他用户的活动检查某些数据对象的相似性的方法，它使我们能够根据该数据做出预测。

在我们的例子中，我们将隐式地从客户群收集购物车/订单数据，我们将使用它来提出更好的产品推荐，以增加在线销售。

执行

首先，我们来看一下数据收集。向购物“下订单”功能添加额外的服务调用并不是什么大不了的事情。事实上，它已经存在了；只是数据被存储在数据库中并稍后进行处理。不要误会：我们仍然希望包括批处理。

但我们还希望实时处理购物车数据，以便我们可以将其直接反馈到在线数据集中并随后立即使用。

我们将首先引入一个事件流解决方案，例如阿帕奇脉冲星。这样，所有新的购物车活动都会放在脉冲星主题，它被消耗并发送到底层批处理数据库，并帮助训练我们的实时 ML 模型。

对于后者，我们的 Pulsar 消费者将写入 Cassandra 表（如图 2 所示），该表仅用于保存订单中每个产品的条目。然后，该产品有一行包含该订单和其他订单中的所有其他产品：

 CREATE TABLE order_products_mapping ( id text, added_product_id text, cart_id uuid, qty int, PRIMARY KEY (id, added_product_id, cart_id) ) WITH CLUSTERING ORDER BY (added_product_id ASC, cart_id ASC); 

然后，我们可以在该表中查询特定产品（本例中为“DSH915”），如下所示：

 SELECT added_product_id, SUM(qty) FROm order_products_mapping WHERE id='DSH915' GROUP BY added_product_id; added_product_id | system.sum(qty) ------------------+----------------- APC30 | 7 ECJ112 | 1 LN355 | 2 LS534 | 4 RCE857 | 3 RSH2112 | 5 TSD925 | 1 (7 rows)

然后我们可以将前四个结果放入产品推荐表中，以供推荐服务通过“ product_id ”进行查询：

 SELECT * FROM product_recommendations WHERE product_id='DSH915'; product_id | tier | recommended_id | score ------------+------+----------------+------- DSH915 | 1 | APC30 | 7 DSH915 | 2 | RSH2112 | 5 DSH915 | 3 | LS534 | 4 DSH915 | 4 | RCE857 | 3 (4 rows)

这样，新的推荐数据就会不断更新。此外，上述所有基础设施资产都位于本地数据中心。

因此，从订单中提取产品关系、通过 Pulsar 主题发送并将其处理为存储在 Cassandra 中的推荐的过程只需不到一秒的时间。

借助这个简单的数据模型，Cassandra 能够在个位数毫秒内提供所请求的建议。

结论和后续步骤

从长远来看，我们希望确保检查数据如何写入 Cassandra 表。通过这种方式，我们可以提前解决与未绑定行增长和就地更新等相关的潜在问题。

可能还需要添加一些额外的启发式过滤器，例如“不推荐”列表。

这是因为有些产品我们的客户只会购买一次或不经常购买，推荐它们只会占用他们更有可能冲动购买的其他产品的空间。

例如，推荐从我们的电器部门购买洗衣机等商品不太可能产生“冲动购买”。

未来的另一个改进是实施像Kaskada这样的实时 AI/ML 平台来处理产品关系流并直接向服务提供推荐数据。

幸运的是，我们确实想出了一种方法来增强现有的、缓慢的批处理过程，使用 Pulsar 来提供要实时处理的购物车添加事件。一旦我们了解了该系统的长期运行情况，我们就应该考虑关闭旧的批处理过程。

PE 承认我们在新解决方案方面取得了良好进展，而且更好的是，我们还开始为消除一些技术债务奠定基础。最后，大家都感觉很好。

在即将发表的文章中，我们将研究如何通过矢量搜索改进产品促销。

了解 DataStax 如何支持实时 AI

作者：Aaron Ploetz，DataStax

也发布在这里