幂等流或使用PipelineDB防止重复行

Question

我的应用程序生成包含多个应用程序度量的旋转日志文件。日志文件每分钟旋转一次，但每个文件仍然相对较大(超过30 of，有100 k行)。

我想把日志输入到PipelineDB (运行在同一台机器上)，Countiuous可以为我创建精确的集合，而不是指标。

我可以使用stdin的副本轻松地将日志发送到PipelineDB，这非常有用。

但是，在日志文件的复制过程中，机器可能会意外地关闭(例如，由于电源不足)。这意味着，一旦恢复联机，就会有多少文件被插入到PipelineDB中存在不确定性。

在这种情况下，如何确保正确插入日志中的每一行？(这是非常重要的，我得到完整和准确的聚合)

注意，日志文件中的每一行都有一个唯一的标识符(由我的应用程序创建的序列号)，但是我在docs中找不到在流中定义唯一字段的选项。我假设管道in的设计并不意味着处理流行中的唯一字段

然而，对这一问题是否有其他解决办法？

Answer 1

流(无限行)上下文中的精确一次语义是一个非常复杂的问题。大多数大型PipelineDB部署都在PipelineDB前面使用某种消息总线基础设施(例如Kafka)来传递语义和可靠性，因为这不是管道‘s的核心关注点。

话虽如此，这里有几种方法是值得考虑的。

首先，您可以在PipelineDB中维护一个常规表，该表跟踪每个日志文件及其成功写入PipelineDB的行号。当开始发送一个新的日志文件时，请在此表中检查它，以确定从哪个行开始。

其次，您可以通过日志文件(通过在分组中包含路径或其他东西)分隔聚合，并在发送之前简单地为该日志文件的任何现有行进行DELETE。然后使用合并在读取时对所有日志文件进行聚合，可能使用VIEW。