标签: event-stream-processing

也许您熟悉事件流处理(ESP)的概念......如果您愿意,我很乐意听到您正在使用的内容以及您正在使用它们的平台.我是Esper项目的积极贡献者(http://esper.codehaus.org/),但我有兴趣听听其他人使用的内容吗？有人使用Coral8,Aleri或Streambase吗？你在哪些平台上使用它们？

我正在设计一个系统,其中有一个主要的对象流,并且有多个工人从该对象产生一些结果.最后,有一些特殊/独特的工作者(根据图论)的某种"接收器",它获取所有结果,并将它们处理成写入某个DB的最终对象.

工人可能依赖于其他一些工人的结果(因此,等待他们的结果)

现在,我面临几个问题:

1. 可能是一名工人比另一名工人慢得多.你怎么处理那件事呢？添加更慢类型的更多工作者(=缩放)？(也许是动态的)
2. 假设W_B依赖于W_A.如果W_B由于某种原因而关闭,则流程将停止,系统将停止工作.所以我想让系统以某种方式绕过这个工人.
3. 此外,最终工作人员如何决定何时对结果进行操作？假设它有A和B的结果,但缺少C的结果.可能是C下降或者此刻它只是非常慢.它怎么能做出决定？

值得一提的是,它不是实时应用程序,而是离线处理系统(即您可以访问数据库并更改记录),但与此同时,它必须以"高速度"处理相对大量的对象".

关于技术,
我正在使用Java开发系统,但我并没有受到特定技术的限制.

如果你能帮助我完成系统的总体设计,我会很高兴的.

非常感谢!

我想处理通过 MQTT 接收到的事件流。我正在使用的库使用回调来提供结果。我正在做的处理取决于以前的状态，而不仅仅是最新的事件。此外，未来的事件可能会从其他来源收集。

一开始我决定把它组合成一个听起来是个好主意的清单。我有一个小问题，因为 IO 阻止了延迟评估并且等待无限流可能很长，但我用交错 IO 解决了它。

stream :: IO [Event]允许我做的不错的东西一样foldl，foldM map，mapM，等...不幸的是这种方法我宁愿将无法在两个流合并，原因是没有更多的锁定功能存在。

我正在挖掘许多库，例如找到了带有 TQueue 的 STM。不幸的是，这不是我真正想要的。

我决定创建自定义类型并制作它，Foldable以便我能够折叠它。由于 IO，我失败了。

import Control.Concurrent.STM

newtype Stream a = Stream (STM a)

runStream
  :: ((a -> IO ()) -> IO i)
  -> IO (Stream a)
runStream block = do
  queue <- newTQueueIO
  block (atomically . writeTQueue queue)
  return $ Stream (readTQueue queue)

foldStream :: (a -> b -> IO b) -> b -> Stream a …

在设计流处理管道时，如果我有许多主题，这些主题至少有一个分区，但可能没有数据进入其中，可能会产生什么成本？

举个例子，对于一个消费者，我可以选择一个“大型主题”，其中包含所有数据和许多分区，或者我可以选择将该数据（按租户、帐户或用户等）拆分为许多主题，默认情况下，单个分区。我对第二种情况的担心是，会有很多主题/分区看不到数据。那么，这个未使用的分区是否会产生任何成本，或者未使用的主题是否会产生任何成本。

我在流中获取字符串数据，我只想保留字母数字字符。我注意到 Siddhi 提供了一个 regexp 函数，正如这里提到的。但问题是它返回一个布尔值而不是修改后的字符串。有没有办法直接获取修改后的字符串？这是我的代码。

@App:name("strtest")
@App:description("Description of the plan")

-- Please refer to https://docs.wso2.com/display/SP400/Quick+Start+Guide on getting started with SP editor. 

define stream InboundStream(ipstring string);

@sink(type='log', prefix='Modified string')
define stream Opstream(ropstring bool);

from InboundStream 
select str:regexp(ipstring, "^A-Za-z0-9") as ropstring insert into Opstream;

是否有返回修改后的正则表达式字符串的函数？