问题标签 [streaming-analytics]

我正在重新设计一个基于流式 IoT 传感器数据的实时预测管道。该管道正在摄取传感器数据样本，其结构与(sensor_id, timestamp, sample_index, value)在源系统中创建时一样，将它们保存在本地并运行 pyspark 批处理作业以训练算法和进行预测。

目前，传感器数据保存在磁盘上的本地文件中，每个传感器只有一个文件，并保存到 HDFS 以进行火花流式传输。流式作业获取每个微批次，计算每个传感器到达的样本数量，并决定哪些传感器积累了足够的新数据来做出新的预测。然后，它将 RDD 中的每个传感器行映射到使用 pythonopen方法打开数据文件的方法，扫描到最后处理的样本，从该样本中提取数据以及预测所需的一些历史数据，然后运行预测作业在火花簇上。此外，每个算法的每个固定数量的样本都需要重新调整，它从同一数据存储中查询很长的历史并在 Spark 集群上运行。

最后，预测作业处理的 RDD 如下所示：

我们现在在监控几十万个传感器时遇到了规模问题。在这个过程中，成本最高的操作似乎是从文件中读取数据——读取每个文件的几十毫秒延迟累积到整个预测作业的无法控制的延迟。此外，将数据作为平面文件存储在磁盘上根本无法扩展。

我们正在研究改变存储方法以提高性能并提供可扩展性。使用时间序列数据库（我们尝试过 timescaledb 和 influxdb）会导致在一个查询中查询所有传感器的数据的问题，当每个传感器需要从不同的时间点查询时，然后将单独的样本分组到sensor_data列中，如图所示以上，这是非常昂贵的，会导致大量的洗牌，甚至不如平面文件解决方案。我们也在尝试 parquet 文件，但是它们的单一写入行为使得计划在这种情况下表现良好的数据结构变得困难。

tl;dr - 我正在为以下场景寻找高性能架构：

1. 实时摄取流式传感器数据
2. 当传感器累积足够的样本时，查询当前+历史数据并将其发送到预测作业
3. 每个预测作业处理在最后一个微批次中达到阈值的所有传感器
4. RDD 包含传感器 ID 行和所有查询样本的有序数组

假设所有事件都准时到达并且不允许迟到，我该如何仅在会话窗口结束时进行一些处理？即水印已通过（lastEventInWindowTimestamp + inactivityGap）。发生这种情况时，我找不到任何调用的 API 方法。我可以使用自定义实现这个逻辑ProcessWindowFunction吗？

在我在 AWS 上的架构中，我有一个运行在 EC2 实例上的服务，它调用 Twitter 流 API 进行数据摄取，即实时推文的摄取。我将此服务称为 TwitterClient。

Twitter API 使用一种基于 HTTP 协议的长轮询来传递流数据。文档说——在您的应用程序（在我的例子中是 TwitterClient）和 API 之间打开了一个连接，并通过该连接发送新的推文。

TwitterClient 然后将实时推文传递到后端（使用 Kinesis Data 流）进行处理。

我面临的问题是 - 并行运行多个 EC2 实例将导致重复的推文被摄取，并且每条推文都将被处理多次。但是，只有一个 EC2 实例会成为单点故障。

我无法承受停机时间，因为我不能错过任何一条推文。

我应该怎么做才能确保高可用性？

编辑：添加了 Twitter API 如何传递流数据的简要说明

我在“Sub1”中有一个事件中心 EH1，在“Sub2”中有一个 Streaming Job ASA1。EH1 是 ASA1 的流输入。为了增加安全性，我计划使用托管身份从流分析作业访问事件中心。 https://docs.microsoft.com/en-us/azure/stream-analytics/event-hubs-managed-identity

为此，必须将“事件中心数据接收器”角色分配给事件中心中的作业。现在由于流式传输作业在另一个订阅中，我无法在添加角色时查找作业。

该解决方案如何进行？Sub2 中的流式传输作业如何在 Sub1 中可见？

感谢您的回复。