Chapter 9. Streaming Joins（流式加入）

当我第一次开始学习联接时，这是一个令人生畏的话题。左，外，半，内，交叉：联接的语言富有表现力和广泛性。最重要的是，流式处理要花费的时间很长，剩下的似乎是一个具有挑战性的复杂主题。好消息是，加入并不是真正令人恐惧的野兽，它们最初看上去似乎是令人讨厌，尖锐的牙齿。与许多其他复杂主题一样，在您理解了连接的中心思想和主题之后，基于这些基础知识的更广泛的领域突然变得更加容易被使用。因此，在我们探索…嗯，加入的迷人话题时，请现在加入我。

您的所有联接都属于流式传输

联接两个数据集意味着什么？我们直观地理解联接只是一种特定的分组操作类型：通过将共享某些属性（即密钥）的数据联接在一起，我们将一些以前不相关的单个数据元素收集在一起，成为一组相关的元素。正如我们在第6章了解到的那样，分组操作总是消耗一个流并产生一个表。知道了这两件事，得出构成整个章节基础的结论只是一个小小的飞跃：在他们心中，所有联接都是流联接。

这个事实的妙处在于，它实际上使流连接的话题变得更加容易处理。我们在流分组操作（窗口，水印，触发器等）的上下文中了解到的所有用于推理时间的工具，在流连接的情况下仍将继续适用。可能令人生畏的是，将流添加到混合中似乎只能使事情复杂化。但是，正如您将在以下示例中看到的那样，将所有联接建模为流联接具有一定的优雅性和一致性。显然，几乎所有类型的连接实际上都归结为相同模式的细微变化，而不是感觉到会有很多不同的连接方法。最终，这种洞察力的清晰有助于降低连接（流式传输或其他方式）的威慑力。

为了给我们提供一些具体的理由，让我们考虑多种不同类型的联接，因为它们应用了以下数据集，方便地命名为Left和Right以匹配常用术语：

每个包含三列：

数词
一个数字。

ID
相应表名称中的第一个字母（“ L”或“ R”）和数字的组合，从而提供了一种在连接结果中唯一标识给定单元格来源的方式。

时间
给定记录在系统中的到达时间，这在考虑流连接时很重要。

为简单起见，请注意我们的初始数据集将具有严格唯一的连接键。当谈到SEMI联接时，我们将介绍一些更复杂的数据集，以在存在重复键的情况下突出显示联接行为。

我们首先深入地研究非窗口式联接，因为窗口化通常仅以较小的方式影响联接语义。在我们厌倦了对未成行的联接的需求之后，我们在窗口化的上下文中触及了联接的一些更有趣的观点。

无窗口加入

流行的神话是，流结合无边界数据总是需要窗口化。但是，通过应用我们在第6章中学到的概念，我们可以发现事实并非如此。连接（有窗口和无窗口）只是分组操作的另一种类型，分组操作产生表。因此，如果我们想将非窗口化联接（或等效地，覆盖整个时间的单个全局窗口中的联接）创建的表作为流使用，则只需要应用非分组化（或触发）操作即可。“等待，直到我们看到所有输入”变体。将联接展开到非全局窗口中并使用水印触发器（例如，“等待直到看到流中有限的时间块中的所有输入”触发器）确实是一个选项，但是在每个触发器上触发记录（即，物化视图的语义）或随着处理时间的增长而定期记录，而不管是否对连接进行了窗口化。因为它使示例易于理解，所以我们假定在以下所有非窗口式连接示例中使用隐式默认每条记录触发器，这些示例以流的形式观察连接结果。

现在，加入自己。ANSI SQL定义了五种连接类型：FULL OUTER，LEFT OUTER，RIGHT OUTER，INNER和CROSS。我们将深入研究前四个，并在下一段中仅简要讨论后四个。我们还将介绍另外两个有趣的但很少见的（并且使用最少的支持，至少使用标准语法）变体：ANTI和SEMI联接。

从表面上看，这听起来像是很多变化。但是，正如您将看到的，核心实际上只有一种类型的联接：FULL OUTER联接。CROSS联接只是FULL OUTER联接，带有虚假的true联接谓词；也就是说，它返回左表中的一行与右表中的一行的所有可能的配对。所有其他所有的联接变体都可以简化为FULL OUTER联接的某些逻辑子集。1因此，在了解了所有不同联接类型之间的共性之后，将它们全都放在头脑中变得容易得多。这也使得在流传输的上下文中对它们进行推理变得非常简单。

开始之前的最后一点说明：我们将主要考虑基数最多为1：1的等值连接，这是指其中连接谓词为等式声明且每一侧最多有一个匹配行的连接的加入。这样可以使示例简单明了。关于SEMI联接时，我们将扩展示例以考虑具有任意N：M基数的联接，这将使我们观察到更多任意谓词联接的行为。

完整外衣

因为它们构成了其他每个变体的概念基础，所以我们首先来看一下FULL OUTER联接。外连接体现了对“连接”一词的相当开放和乐观的解释：FULL OUTER-连接两个数据集的结果本质上是两个数据集中的行的完整列表，2两个数据集中共享相同连接的行 键组合在一起，但任一侧的不匹配行均未连接。

例如，如果我们将FULL OUTER完全连接起来-将我们的两个示例数据集连接到仅包含连接ID的新关系中，结果将类似于以下内容：

我们可以看到，FULL OUTER连接包括满足连接谓词的两行（例如，“ L2，R2”和“ L3，R3”），但是它还包含部分谓词失败的行（例如，“ L1， null”和“ null，R4”，其中null表示数据的未连接部分。

当然，这只是此FULL OUTER-join关系的时间点快照，是在所有数据到达系统后拍摄的。我们是在这里学习流式连接的，而流式连接的定义涉及时间的增加。从第8章我们知道，如果我们想了解给定的数据集/关系如何随时间变化，我们想说的是时变关系（TVR）。因此，为了最好地了解联接随着时间的变化，现在让我们看一下该联接的完整TVR（每个快照关系之间的变化以黄色突出显示）：

而且，正如您可能期望的那样，此TVR的流呈现将捕获每个快照之间的特定增量：

请注意，其中包含“时间”和“撤消”列，以突出显示给定行在流中具体化的时间，并在对给定行进行更新首先导致撤回该行的先前版本时调出实例。如果此流要捕获一段时间内TVR的全保真视图，则撤消/撤消行至关重要。

因此，尽管联接的这三个呈现（表，TVR，流）中的每个呈现都彼此不同，但也很清楚它们是如何在同一数据上只是不同的视图：表快照向我们显示了整体在所有数据到达之后就存在的数据集，TVR和流版本（以自己的方式）捕获整个关系在其存在过程中的演变。

有了FULL OUTER连接的基本知识，我们现在了解了流上下文中连接的所有核心概念。无需开窗，无需定制触发器，也无需特别痛苦或不直观的操作。正如您所期望的，只是随时间变化的连接的每条记录的演变。更好的是，所有其他类型的联接都只是该主题的变体（至少在概念上是这样），本质上仅是对FULL OUTER联接的每个记录流执行的附加过滤操作。现在让我们更详细地研究它们中的每一个。

左外

LEFT OUTER联接只是FULL OUTER联接，已删除了右数据集中所有未联接的行。通过采用原始的FULL OUTER连接并将要过滤的行显示为灰色，可以最清楚地看到这一点。对于LEFT OUTER联接，如下所示，其中左侧未联接的每一行均从原始FULL OUTER联接中滤除：

要查看表和流在实际中的实际外观，让我们再次查看相同的查询，但是这次将灰色行完全省略：

正确的外表

RIGHT OUTER联接与左联接相反：完全外部联接中来自左数据集的所有未联接行都被清除，咳嗽，已删除：

在这里，我们看到呈现为实际RIGHT OUTER连接的查询将如何出现：

内

INNER联接本质上是LEFT OUTER和RIGHT OUTER联接的交集，或者，以减法考虑，从原始FULL OUTER联接中删除的行创建一个INNER联接是从LEFT OUTER和RIGHT删除的行的并集 OUTER加入。结果，INNER联接中缺少任一侧未联接的所有行：

再一次，INNER联接的渲染更加简洁：

在此示例中，您可能会倾向于认为缩进永远不会在INNER连接流中发挥作用，因为在此示例中，所有缩进都被过滤掉了。但是，假设是否在12:07在“左”表中将Num为3的行的值从“ L3”更新为“ L3v2”。除了在最终的TABLE查询的左侧产生不同的值（再次在12:10进行，这是在对左侧的第3行进行更新之后）之外，这还将导致STREAM捕获两个 通过撤消和添加新值来删除旧值：

反对

反对

反对联接与INNER联接相反：它们包含所有未联接的行。并非所有的SQL系统都支持干净的ANTI连接语法，但是为了清楚起见，这里我将使用最直接的语法：

关于ANTI连接的流呈现的一点点有趣之处在于，它最终包含了一堆错误的开始和缩回，导致行最终最终合并。实际上，ANTI联接的缩进量与INNER联接的缩量一样重。更简洁的版本如下所示：

半

现在我们来谈谈SEMI联接，而SEMI联接有点奇怪。乍一看，它们基本上看起来像内部联接，而联接值的一侧被删除了。而且，实际上，在：的基数关系为N：M且M≤1的情况下，这是可行的（请注意，我们将使用keep = Left， drop =对后面的所有示例都是正确的）。例如，到目前为止，在我们使用的Left和Right数据集上（连接数据的基数分别为0：1、1：0和1：1），INNER和SEMI连接变体看起来是相同的：

但是，如果M：1的N：M基数，则SEMI联接还有一个细微之处，因为未返回M侧的值，所以SEMI联接只是在存在任何匹配行的情况下确定联接条件。正确，而不是为每个匹配的行重复产生新的结果。
为了清楚地看到这一点，让我们切换到一对稍微复杂一些的输入关系，以突出显示其中包含的行的N：M连接基数。在这些关系中，N_M列说明行的基数关系在左侧和右侧之间，而Id列（如前所述）为每个输入关系中的每一行提供唯一的标识符：

使用这些输入，FULL OUTER联接将扩展为如下所示：

附带说明一下，这些更复杂的数据集的另一个好处是，当每一侧上有多个行与同一谓词匹配时，联接的乘法性质开始变得更加清晰（例如，“ 2：2”行，它们会扩展 从每个输入的两行到输出的四行；如果数据集具有一组“ 3：3”行，则它们将从每个输入的三行扩展到输出的九行，依此类推 ）。

但是回到SEMI的精妙之处。使用这些数据集，可以更清楚地了解过滤后的INNER联接和SEMI联接之间的区别是：INNER联接产生N：M基数M> 1的任何行的重复值，而SEMI联接则不会。t（请注意，我已用红色突出显示了INNER联接版本中的重复行，并以灰色突出显示了在各个INNER和SEMI版本中省略的完整外部联接的部分）：

或者，更简洁地呈现：

然后，STREAM渲染提供了一些关于哪些行被滤除的上下文-它们只是稍后到达的重复行（从所投影列的角度来看）：

再次，简洁地呈现：


正如我们在许多示例过程中看到的那样，流连接确实没有什么特别的。鉴于我们对流和表的了解，它们的功能完全符合我们的期望，并且随着连接流的发展，连接流会捕获连接的历史记录。这与联接表相反，联接表只是捕获特定时间点存在的整个联接的快照，这可能是我们习惯了。

但是，更重要的是，通过流表理论的视角进行的观看联接提供了一些额外的清晰度。底层基础联接原语的核心是FULL OUTER联接，它是流→表分组操作，它将关系中的所有联接和未联接的行收集在一起。我们详细研究过的所有其他变体（LEFT OUTER，RIGHT OUTER，INNER，ANTI和SEMI）仅在FULL OUTER连接之后在连接的流上添加了一层额外的过滤。

窗口联接

在查看了各种非窗口式连接之后，接下来让我们探索加窗功能。我会争辩说，有两种动机可以使您的加入窗口化：

以有意义的方式分配时间
一个明显的例子是固定的窗户。例如，出于某些业务原因（例如，每天的帐单记录），应将在同一天发生的事件的每日窗口合并在一起。出于性能原因，另一个可能是限制联接内的时间范围。但是，事实证明，存在更多复杂的（并且有用的）连接时间划分方式，包括一个特别有趣的用例，即我今天所知的任何流系统都不是本地支持的：时间有效性连接。只需一点更多的信息。

为超时联接提供有意义的参考点
这对于许多无限制的连接情况很有用，但是对于外部连接之类的用例来说，这可能是最明显的好处，对于这些用例来说，连接的一侧是否会出现是未知的。对于经典的批处理（包括标准的交互式SQL查询），仅当边界输入数据集已被完全处理时，外部联接才会超时。但是，在处理无限制数据时，我们不能等待所有数据都被处理。正如我们在第2章和第3章中讨论的那样，水印为衡量事件时间中输入源的完整性提供了一种进度指标。但是要使用该度量来超时联接，我们需要一些参考点进行比较。窗口化联接通过将联接的范围限制到窗口的末尾来提供该引用。水印通过窗口末尾后，系统可能会认为该窗口的输入已完成。到那时，就像在有界联接情况下一样，使任何未联接的行超时并实现其部分结果是安全的。

就是说，正如我们前面所看到的，窗口连接绝对不是流连接的必要条件。在很多情况下，这很有道理，但绝不是必须的。

实际上，大多数窗口联接的用例（例如，日常窗口）都是相对简单明了的，并且很容易从到目前为止我们学到的概念中推断出来。为了了解原因，我们简要地介绍一下将固定窗口应用于我们已经遇到的一些连接示例的含义。此后，我们将在本章的其余部分中研究时间有效性联接的更有趣（且更加弯曲）的主题，首先详细研究我所指的时间有效性窗口，然后继续研究什么连接是指在此类窗口的上下文中。

固定视窗

窗口化联接将时间维度添加到联接条件本身中。在这种情况下，窗口的作用是将只连接到窗口时间间隔内的行集合的作用域限定为范围。也许可以通过一个示例更清楚地看到这一点，所以让我们将原始的左右表放入五分钟的固定窗口中：

在我们之前的Left和Right示例中，连接条件只是Left.Num = Right.Num。要将其变为窗口式联接，我们将联接条件扩展为也包括窗口相等性：Left.Num = Right.Num AND Left.Window = Right.Window。知道了这一点，我们已经可以从前面的窗口表中推断出联接的变化方式（为清楚起见而突出显示）：由于L2和R2行不在同一固定的五分钟内，因此它们不会在 我们加入的窗口变体。

实际上，如果我们将未窗口化和窗口化的变体并排比较为表格，我们可以清楚地看到这一点（在连接的每一侧都突出显示了对应的L2和R2行）：

当比较未窗口化和窗口化的连接作为流时，这种差异也很明显。正如我在以下示例中突出显示的那样，它们的主要区别在于它们的最后一行。无窗口的一侧完成了Num = 2的联接，除了为已完成的L2的新行R2联接之外，对于未联接的R2行也产生了退缩。另一方面，窗口化的一侧仅产生未连接的L2行，因为L2和R2落在不同的五分钟窗口之内：


至此，我们现在了解了窗口对FULL OUTER连接的影响。通过应用我们在本章上半部分中学到的规则，可以轻松导出左外，右外，内，反和SEMI连接的窗口化变体。我将把大多数派生作为练习留给您完成，但举一个例子，据我们了解，LEFT OUTER连接只是FULL OUTER连接，而连接左侧的空列被删除了（再次， 突出显示L2和R2行以比较差异）：

通过将连接的时间范围划分为固定的五分钟间隔，我们将数据集分为两个不同的时间窗口：[12:00，12:05）和[12:05，12:10）。然后，在这些区域中应用了我们之前观察到的完全相同的连接逻辑，对于L2和R2行落入单独区域的情况，产生的结果略有不同。从根本上讲，这就是窗口联接的全部内容。

时间有效性

看完了窗口连接的基础知识之后，我们现在将本章的其余部分花在一种更高级的方法上：时间有效性窗口。

时间有效性窗口

时间有效性窗口适用于以下情况：关系中的行有效地将时间分割为给定值有效的区域。更具体地说，想象一个用于执行货币兑换的金融系统。4这种系统可能包含随时间变化的关系，该关系捕获了各种货币的当前汇率。例如，可能存在将不同货币转换为日元的关系，如下所示：

为了突出我的意思，我说的是时间有效性窗口“将时间有效地分割为一个给定值有效的区域”，请仅考虑该比率中的欧元兑日元转换率：

从数据库工程的角度来看，我们理解这些值并不意味着将欧元精确转换为日元的汇率在12:00时为114¥/€，在12:03时为116¥/€，在12时为119¥/€： 06，其他所有时间均未定义。取而代之的是，我们知道此表的目的是要捕捉一个事实，即直到12:00，从12:00到12:03的¥114 /€，从12到116的€/€为止，欧元到日元的转换率是不确定的： 03至12:06，从那时起119¥/€。或在时间表中列出：

现在，如果我们提前知道所有速率，就可以在行数据本身中明确捕获这些区域。但是，如果我们只需要基于给定速率生效的开始时间来逐步建立这些区域，那么我们就会遇到一个问题：给定行的区域会随着时间而变化，具体取决于之后的行 它。即使数据按顺序到达也是一个问题（因为每次到达新速率时，先前的速率从永远有效变为有效，直到新速率到达时间），但是如果它们可以到达，则情况更加复杂。的顺序。例如，使用前面的YenRates表中的处理时间顺序，我们的表随时间推移将有效表示的时间线序列如下：

或者，如果我们想将此呈现为随时间变化的关系（每个快照关系之间的变化以黄色突出显示）：

这里要注意的是，一半的更改涉及对多行的更新。听起来似乎还不错，直到您回想起每个快照之间的区别是恰好有一个新行到达。换句话说，单个新输入行的到来导致对多个输出行的事务性修改。听起来不太好。另一方面，这听起来也很像建立会话窗口中涉及的多行事务。的确，这是窗口化的又一个示例，它提供了超越简单的时间划分的好处：它还提供了以涉及复杂的多行事务的方式实现此目的的能力。

要查看实际效果，让我们看一个动画。如果这是Beam管道，则可能类似于以下内容：

PCollection<Currency, Decimal> yenRates = ...; PCollection<Decimal> validYenRates = yenRates .apply(Window.into(new ValidityWindows())
.apply(GroupByKey.<Currency, Decimal>create());

该管道以流/表动画的形式呈现，如图9-1所示。

图9-1 时间效度窗口化

该动画突出了时间有效性的一个关键方面：缩小窗口。有效性窗口必须能够随着时间的推移而缩小，从而减小其有效性的范围，并将其中包含的任何数据分配到两个新窗口中。有关部分实现的示例，请参见GitHub上的代码片段。

用SQL术语来说，这些有效性窗口的创建类似于以下内容（使用假设的VALIDITY_WINDOW构造），被视为表：

或者，也许更有趣的是，将其视为流：

太好了，我们对如何使用时间点值有效地将时间切分成有效的范围有所了解。但是，这些时间有效性窗口的真正威力在于将它们与其他数据结合起来时的应用。这就是时间有效性加入的地方。

时间有效性加入

为了探索时间有效性联接的语义，假设我们的金融应用程序包含另一个时变关系，该关系跟踪从各种货币到日元的货币转换顺序：

为简单起见，和以前一样，我们将重点放在欧元换算上：

我们希望将这些订单牢固地加入到YenRates关系中，并将YenRates中的行视为定义有效期窗口。因此，我们实际上要加入到上一节末尾构造的YenRates关系的有效性窗口版本：

幸运的是，在将转换率放入有效性窗口之后，这些转换率与YenOrders关系之间的窗口式联接为我们提供了我们想要的：


回想我们最初的YenRates和YenOrders关系，这种连接关系的确看起来是正确的：对于给定的事件时间窗口，三个转换中的每个转换都以（最终）适当的速率结束，在该时间范围内它们的对应顺序下降。因此，我们有一种体面的感觉，即这种联接在为我们提供最终想要的正确性方面正在做我们想要的事情。

就是说，在所有值到达并且尘埃落定之后获取的这种简单的关系快照视图掩盖了此连接的复杂性。要真正了解此处发生的情况，我们需要查看完整的TVR。首先，请记住，有效性窗口转换率关系实际上比以前的简单表快照视图复杂得多，您可能会相信。作为参考，以下是有效期窗口关系的STREAM版本，该版本更好地突出了这些转换率随时间的变化：

结果，如果我们查看有效性窗口连接的完整TVR，您会发现该连接随时间的演变要复杂得多，这是由于值的无序到达 联接：


特别是，5欧元订单的结果最初报价为570日元，因为该订单（发生在12:05）最初属于114日元/欧元汇率的有效期窗口。但是，当活动时间12:03的116¥/€的价格出现混乱时，则5€订单的结果必须从570¥更新为580¥。如果您以流的形式观察连接的结果，这也很明显（在这里，我用红色突出显示了错误的570¥，用蓝色突出显示了570¥的撤消操作，然后用蓝色突出显示了580¥的更正值）：

值得一提的是，由于使用了FULL OUTER连接，这是一个相当混乱的流。实际上，当将转换订单作为流使用时，您可能并不关心未连接的行； 切换到INNER联接有助于消除这些行。您可能也不在乎汇率窗口发生变化的情况，但实际转化价值不会受到影响。通过从流中删除速率窗口，我们可以进一步减少其闲聊：

好多了。现在，我们可以看到该查询非常简洁地执行了我们最初打算执行的操作：以两个健壮的方式合并两个TVR以获取货币转换率和订单，从而可以容忍数据无序到达。图9-2将该查询可视化为动画图。在其中，您还可以非常清楚地看到事物整体结构随时间变化的方式。

图9-2 时间有效性联接，通过每次记录触发将欧元转换为日元

水印和时间有效性结合在一起。在此示例中，我们重点介绍了在本节开始部分提到的窗口化联接的第一个好处：窗口化联接使您可以根据实际业务需要在一定时间内对联接进行分区。在这种情况下，业务需求是将时间划分为适用于我们的货币兑换率的有效区域。

但是，在我们称之为“一天”之前，事实证明该示例还提供了一个机会来突出显示我提到的第二点：加窗连接可以为水印提供有意义的参考点这一事实。为了了解它的用处，想象一下改变前一个查询，用显式水印触发器替换隐式默认的每条记录触发器，该显式水印触发器仅在水印通过联接中的有效性窗口的末尾时才触发一次（假设我们有我们的两个输入TVR都提供了水印，可以准确地跟踪事件时间中这些关系的完整性，并且执行引擎知道如何考虑这些水印）。现在，我们的流将不再包含多个因乱序到达而导致的支出和撤回，而是可以最终得到一个流，其中包含每个定单的正确转换结果，这显然比以前更加理想：


或者，以动画的形式呈现，它清楚地显示了合并的结果如何直到水印移到输出流之后才被输出到输出流中，如图9-3所示。

图9-3 时间有效性加入，通过水印触发将欧元转换为日元

无论哪种方式，看到此查询如何将如此复杂的交互集合封装为所需结果的简洁明了的呈现方式，都令人印象深刻。

概括

在本章中，我们在流处理的上下文中分析了连接的世界（使用SQL的连接词汇）。我们从无窗口联接开始，从概念上看，所有联接如何以流联接为核心。我们看到了基本上所有其他联接变体的基础是FULL OUTER联接，并讨论了在LEFT OUTER，RIGHT OUTER，INNER，ANTI，SEMI甚至CROSS联接的一部分中发生的特定更改。另外，我们看到了所有这些不同的联接模式如何在TVR和流的世界中相互作用。

接下来，我们进入了窗口联接，并了解了使窗口联接通常是受以下一项或两项好处的推动：

• 能够在一定时间内根据业务需求对联接进行分区
• 将连接结果与水印进度联系起来的能力

最后，我们深入探讨了关于连接的一种更有趣和有用的窗口类型：时间有效性窗口。我们看到时间有效性窗口如何仅根据那些值改变的特定时间点自然地将时间刻入给定值的有效性区域。我们了解到，有效性窗口内的联接需要一个窗口框架，该框架支持可以随时间推移而拆分的窗口，这是当今现有的流系统都不具备的本机支持。而且，我们看到了有效期窗口如何简洁明了地使我们能够以稳健，自然的方式解决将TVR转换为货币转换率和定单的问题。

联接通常是数据处理，流式传输或其他方面更令人生畏的方面之一。但是，通过了解连接的理论基础以及我们可以直接从该基本基础派生所有不同类型的连接的方式，即使流式传输增加了混合的时间，连接也变得不那么令人恐惧。