Apache Beam for Search：黑客时间入门

要创建相关的搜索，处理点击流数据是关键：您通常希望宣传被点击和购买的搜索结果，并降级用户不喜欢的内容。

通常，搜索系统将处理点击流数据视为处理历史数据的批处理作业，也许使用Spark之类的系统。但是在Shopify的Discovery团队中，我们提出了一个问题：如果用户与搜索结果互动时，我们可以实时自动自动调整相关性，而不必等待几天才能运行大量批处理作业，该怎么办？

在Shopify，这就是我们正在做的！我们正在使用流数据处理系统，该系统可以处理实时数据和历史数据，以实现实时用例，范围包括简单的文档自动增值或降价，计算聚合的点击流行度统计信息，构建离线搜索评估集以及进行更复杂的强化学习任务。

但是本文向您介绍了流系统本身。特别是Apache Beam。最重要的是要考虑使用这些流系统的时间。因此，让我们开始吧！

Apache Beam是一个统一的批处理和流处理系统。这使我们有可能统一一个系统中用户搜索行为的历史和实时视图。 Beam希望能够将这些工作流保持在一起，而不是像Spark这样的批处理系统来处理数月的旧数据，而采用像Apache Storm这样的独立流系统来处理实时用户流量。

对于搜索，这非常令人兴奋。这意味着我们可以构建既依赖于历史搜索日志的搜索系统，又可能能够以各种方式对用户的需求进行实时调整。

让我们经历一下每个人与Beam一起面临的早期挑战：那是时间！ Beam是一种时间机器，在被许多中间处理和存储步骤所烦扰地延迟之后，必须在正确的位置重新排序事件。这是流媒体系统的核心并发症之一-我们要等待多长时间？我们如何处理逾期或故障数据？

因此，要开始使用Beam，首先要做的就是Hack Time！

管道是Apache Beam的核心。他们通过各种处理步骤将源连接到最终的接收器。

流经管道的数据带有时间戳。考虑流媒体系统时，这很有意义。随着事件从浏览器，通过API和其他数据系统的流动，我们会遇到各种延迟。最终，事件到达了我们的Beam管道。它们很容易失序或延迟。光束源API（例如Kafka的API）可维护事件数据的移动视图，以发出井井有条的事件，称为水印。

如果我们没有为Beam来源提供有关如何建立时间戳的良好信息，我们将丢弃事件或以错误的顺序接收事件。但对于搜索而言，更重要的是，我们可能必须组合不同的数据流，以在搜索会话或查询上构建单个视图，如下所示：

加入（另一天是Beam主题！）需要回顾每个来源的水印，并确保它们在时间上对齐，然后再确定是否有足够的时间再继续。但是在您了解串流连接的复杂性之前，以准确的时间戳进行重放是Beam-for-clickstream之旅的第一个里程碑。

让我们建立一个简单的Beam管道来探索Beam。在这里，我们以Java中的Kafka为例。您可以在此要点中查看完整的源代码。

在这里，我们将设置一个Kafka源，这是生成自定义SearchQueryEvent的管道的开始，该自定义SearchQueryEvent存储在search_queries_topic中。

您会注意到，我们拥有有关检索数据的主题/服务器的信息，以及如何反序列化基础二进制数据。我们可能会添加其他处理步骤以转换或处理SearchQueryEvent，最终将最终输出发送到另一个系统。

但是时间还没有。默认情况下，产生的SearchQueryEvents将使用Kafka处理时间。也就是说，当他们从Kafka中阅读时。就我们的目的而言，这是最不有趣的。我们关心的是用户实际何时搜索和点击结果。

更有趣的是在Kafka客户端中创建事件时。我们可以在这里添加：

您会在上方注意到，当我们在下方使用“创建时间”时，我们需要为来源的水印提供提示，以提示事件发生时间可能不正确。例如，下面我们指示Kafka来源使用创建时间，但可能会有5分钟的差异。

让我们来思考一下，从最后一个代码片段开始延迟5分钟实际上意味着什么。 Beam是一台时间机器……Beam如何弯曲时空，这是您的心灵开始受到伤害的地方。

您可能正在接送，活动时间与处理时间完全不同！因此，在上面的代码段中，我们*不是*告诉计算机等待5分钟的执行时间来获取更多数据。不可以，事件时间可能会根据历史数据重播，其中5分钟的事件时间将在短短几毫秒内通过我们的管道重播。或者可能是事件时间已经到了，我们正在积极地传输实时数据以进行处理。因此，我们确实要等待5分钟！

让我们退后一步，并使用一个愚蠢的示例来理解这一点。这对您的Beam旅程至关重要。

想象一下，我们是可以以1000倍速度观看电影的超级机器人机器人。可能就像《星际迷航》的下一代Lt Commander数据一样。如果您不熟悉，他可以像显示屏幕一样快地处理输入！数据可能会说：“嗨，我想看80年代的经典电影《 The Goonies》，所以我可以成为Enterprise员工的文化参考。”

Beam就像在超快进模式下观看电影一样，相对于电影时间中的其他块，视频块可能出现延迟或乱序。在这种情况下，我们有两种时间感：

事件时间：The Goonies的实际1h 55分钟运行时间中的时间戳，也称为电影时间。

处理时间：我们实际体验The Goonies的时间（如果我们是Data之类的超级机器人，则可能只有几分钟）。

因此，Data告诉企业计算机“您可以快快地从存储库中调用Goonies，快来玩。”而且计算机有很多打ic的地方，电影的某些帧并没有完全到达Data的屏幕以保持电影的秩序。

指挥官数据可以容忍丢失这些帧。因此，Data表示：“看，在影片时间*（又称事件时间）中，不要等待超过5分钟，然后才向我展示您在电影中的那部分。这样一来，Data就可以在短时间内观看完整的电影，从而减少了可忍受的电影帧数。

这正是Beam对我们的搜索查询数据所做的。有时它会重播数天的历史搜索数据（以毫秒为单位），而有时我们正在流传输实时数据，我们实际上必须等待5分钟才能处理现实。当然，正确的延迟时间可能不是5分钟，这可能是其他适合我们需要的时间。

Beam具有其他原语，例如窗口，它们可以在源之外进一步告知应如何以时间为单位缓冲或收集数据。我们应该在日常窗口中收集搜索数据吗？我们应该容忍最新数据吗？后续处理需要处理什么？ Windows还可以同时使用计算机概念，而与Beam一起使用时必须深刻理解。

Beam可能对Kafka有所了解，但实际上对我们的数据模型一无所知。有时，我们需要对Beam时间机器中的时间定义进行更多控制。

例如，在我们之前的电影示例中，电影帧可能具有一些字段，可告知我们如何在电影时间安排它们。如果我们检查SearchQueryEvent，还将看到嵌入在数据本身中的特定时间戳：

通常可以将Well Beam源配置为使用自定义事件时间，例如我们的searchTimestamp。我们只需要制定一个TimestampPolicy。我们只需提供一个简单的函数类，该函数类即可获取记录（Long-> SearchQueryEvent的键值）并返回时间戳：

在这里，我们传递了自己的功能，并设置了相同的允许延迟（5分钟）。所有这些都包装在工厂类TimestampPolicyFactory SearchQueryTimestampPolicyFactory中（现在，如果这听起来不像Java类名，我不知道该怎么做；））

Beam是关于黑客时间的，我希望您对Beam的某些功能进行了逐步了解。如果您有兴趣与我一起建立Shopify在搜索和发现方面的未来，请查看这些出色的职位！

Doug Turnbull是Shopify的搜索相关性高级工程师。他以撰写“相关搜索”一书，为“人工智能驱动的搜索”做出贡献以及为Solr和Elasticsearch创建相关工具（例如Splainer，Quepid和Elasticsearch Learning to Rank插件）而闻名。 Shopify的道格（Doug）团队帮助商家提高其产品和品牌的知名度。如果您想与Doug合作，请通过@softwaredoug向他发送推文！

来自构建和扩展Shopify的团队的故事。 这个商业平台为全球超过1,000,000个企业提供支持。