Azure 流分析获取前一个输出行以加入输入

Posted 2023-05-08

【中文标题】Azure 流分析获取前一个输出行以加入输入

【英文标题】：Azure Stream Analytics Get Previous Output Row for Join to Input

【发布时间】：2017-06-08 20:13:34

【问题描述】：

我有以下场景：

移动应用程序生成的事件发送到事件中心，事件中心是流分析查询的输入流源。从那里它们通过顺序查询流传递，该查询根据标准将流分成 2 个流，评估其他条件并决定是否让事件继续流经管道（如果不是，则简单地丢弃） .您可以将我们正在做的事情归类为降噪/事件过滤。基本上，如果 A 刚刚发生，除非 B & C 发生或 X 时间过去，否则不要让 A 再次发生。在查询挑战结束时，流再次合并，“选择”事件作为“选择”输出传播。

我的问题是我需要能够将当前事件与之前的“选择”事件（不仅仅是之前的输入事件）进行比较，所以本质上我需要将我的输入流加入到我的输出流中。我已经尝试了各种方法来做到这一点，但到目前为止都没有奏效，我知道其他 CEP 引擎支持这个概念。我的查询大多都定义为 WITH 语句内的临时结果集（这是我的初始输入流被拉入第一个查询的地方，并且每个后续查询都取决于它上面的查询），但我看不出有办法将我的输入加入到我的输出或将我的输入加入到链中更靠后的另一个临时结果集。好像join只支持输入？

目前，我正在尝试通过一些我真的不想在生产中做的事情来解决这个限制，但我实际上定义了一个输出到 Azure 队列，然后是一个由该队列上的事件触发的 Azure 函数唤醒并将其发布到不同的事件中心，该中心被映射为循环反馈输入，返回到我可以加入的查询中。仍然把所有这些都连接起来，所以不能 100% 确定它会起作用，但认为对于这种相对常见的模式必须有更好的选择？

【参考方案1】：

WITH 语句确实是将先前的输入与其他一些数据连接起来的正确方法。 您可能需要将它与 LAG operator 结合使用，后者会在数据流中获取上一个事件。

让我们知道它是否适合您。

谢谢，

JS - Azure 流分析

【参考方案2】：

AFAIK，流分析作业支持两种不同的数据输入类型：数据流输入和参考数据输入。据我了解，您可以利用 Reference data 来执行查找或与您的数据流相关联。更多详情，您可以参考以下教程：

Data input types: Data stream and reference data

Configuring reference data

Tips on refreshing your reference data

Reference Data JOIN (Azure Stream Analytics)

【讨论】：

谢谢，我知道参考数据，但我担心的是我不明白该机制是为将以非常高的速率变化的数据设计的，这意味着每个入站事件我们的情况最坏的情况。假设我平均每秒发生 20 个事件，其中至少 1/2 的事件最终会更新参考数据，那么参考数据是否可以在这种变化率下使用？

AFAIK，参考数据是静态或随时间缓慢变化的辅助数据。您可以使用samples/date/time/sample.csv 的模式，时间格式为“HH-mm”，指示流分析获取更新的 blob，但时间限制为分钟。如果您的事件数据随时间快速变化，则参考数据输入可能不符合您的场景。