标签: stream-processing

给定一个包含大量项目的结果流，我想存储它们并处理潜在的并发冲突：

public void onTriggerEvent(/* params */) {
  Stream<Result> results = customThreadPool.submit(/*...complex parallel computation on multiple servers...*/).get();
  List<Result> conflicts = store(results);
  resolveConflictsInNewTransaction(conflicts);
}

我被困在如何store(...)有效地实施。在Result由描述数据，需要在其各自的数据库表被更新这两件不更改和拆卸对象。

@Value
public static class Result {
    A a; // describes update for row in table a
    B b; // describes update for row in table b
}

A并且B每个引用两个用户，其中(u1, u2)是各自数据库表上的一个键。

@Value
public static class A {
    long u1;
    long u2;
   // ... computed data fields ...
} …

我正在设计一个原型实时监视器,用于处理相当大量(> 30G /天)的流数字数据.我想在Clojure中写这个,因为该语言似乎非常适合那种"观察者+状态机"系统,这可能最终会成为.

我找到的两个主要候选人是Lamina和Storm.还有Riemann和Pulse,但前者似乎更像是一个完整的解决方案,而不是一个框架,我宁愿不承诺最终的设计; Pulse的回购看起来有点不受维护？

我想知道的是; 这两个项目针对哪些数据和工作流程进行了优化？风暴似乎更成熟,但Lamina似乎更具有组合性和"Clojureic"(我的背景是Python,所以我倾向于高度评价).

我在网上看到的内容:

• Storm似乎是以Big Data(流)为重点,核心是带有Clojure DSL的直接Java.它似乎具有针对许多现有数据源的pre = built处理程序.

• Lamina更像是一个轻量级,可重用的组件,可以对Clojure进行编码以实现抽象,这意味着它可以作为其他事件系统的基础重用.数据源需要在代码中处理.

• 两者都有一组有用的聚合/分裂/计算库函数.Lamina的graphviz集成是一个不错的选择.

在大量使用 Haskell 和函数式语言之后，我开始学习 C++，我发现我一直在尝试解决同样的问题：

• 从输入流中读取一些数据
• 根据特定算法对它们进行标记
• 处理令牌

如果这是 Haskell，我可以简单地利用一切都是懒惰的事实，并在我想到的时候编写我的转换，然后它会在下游被消耗时应用。甚至有一些库可以执行这种精确的模式（导管和管道）。

假设我想获取序列1 2 3 4 5 6 ...和输出12 34 56 ...。我可以了解如何编写在流上运行并就地处理数据的临时代码。但我想知道是否存在一种抽象机制，允许我通过转换来自另一个流的数据（以任何可以想到的方式）来构建新的流。这种抽象应该允许我在处理数据时缓冲数据，而不仅仅是单个元素到新值的简单映射。

以下是限制：

• 除了 stdlib 之外，我无法使用任何其他库。
• 它必须适用于 C++03（意味着没有 C++11 功能。）

如果你在想，这是作业吗？好吧，我收到了很多类作业，这些作业要求我处理数据流（这就是没有库和 C++03 限制的原因）。并不是我不知道如何使用while循环来做到这一点，而是我想知道 stl 中是否存在现有的流抽象，只是等待被发现和使用。

但如果唯一的方法是使用 C++11，那么我想知道。

我正在尝试使用*subFlows连接流.因此,我从广播的出口建立了一个来源.但它抛出了一个UnsupportedOperationException: cannot replace the shape of the EmptyModule.我试图谷歌这个例外,但我找不到类似的东西.

在这里我的代码

val aggFlow = Flow.fromGraph(GraphDSL.create() { implicit builder =>
    val broadcast = builder.add(Broadcast[MonitoringMetricEvent](2))
    val bc = builder.add(Broadcast[Long](1))

    val zip = builder.add(ZipWith[StreamMeasurement, Long, (StreamMeasurement, Long)]((value, ewma) => (value, ewma)))
    val merge = builder.add(Merge[Seq[StreamMeasurement]](1))


    broadcast.out(1) ~> identityFlow ~> maxFlow ~> bc

    val source = Source.fromGraph(GraphDSL.create() { implicit bl =>
      SourceShape(bc.out(0))
    })

    broadcast.out(0) ~> identityFlow ~> topicFlow.groupBy(MAX_SUB_STREAMS, _._1)
        .map(_._2)
        .zip[Long](source)
        .takeWhile(deciderFunction)
        .map(_._1)
        .fold[Seq[StreamMeasurement]](Seq.empty[StreamMeasurement])((seq, sm) => seq:+sm)
      .mergeSubstreams ~>  merge



    FlowShape(broadcast.in, merge.out)
  })

在这里得到的例外情况: …

我想知道在 Flink 中是否可以在运营商之间共享状态。

例如，假设我在操作符上按键进行分区，并且我需要分A区内的一段分区状态C（出于任何原因）（图 1.a），或者我需要C下游操作符中的操作符状态F（图 1 .b)

我知道可以broadcast记录到所有分区。因此，如果您在记录中包含操作符的内部状态，则可以与下游操作符共享您的内部状态。
然而，这可能是一个昂贵的操作，而不是简单地让op1具体请求op2状态。

最近围绕可查询状态的发展是否朝着这个概念发展，或者它们只是为了让外部用户查询拓扑的内部状态？

预先感谢您的见解

在一个大型 json 文件中，我想从嵌套列表中删除一些元素，但保留文档的整体结构。

我的示例将其输入为（但真实的输入足够大以要求流式传输）。

{
  "keep_untouched": {
    "keep_this": [
      "this",
      "list"
    ]
  },
  "filter_this":
  [
    {"keep" : "true"},
    {
      "keep": "true",
      "extra": "keeper"
    } ,
    {
      "keep": "false",
      "extra": "non-keeper"
    }
  ]
}

所需的输出仅删除了“filter_this”块的一个元素：

{
  "keep_untouched": {
    "keep_this": [
      "this",
      "list"
    ]
  },
  "filter_this":
  [
    {"keep" : "true"},
    {
      "keep": "true",
      "extra": "keeper"
    } ,
  ]
}

处理此类情况的标准方法似乎是使用“truncate_stream”来重构流对象，然后再以通常的 jq 方式过滤这些对象。具体来说，命令：

jq -nc --stream 'fromstream(1|truncate_stream(inputs))' 

提供对对象流的访问：

{"keep_this":["this","list"]}
[{"keep":"true"},{"keep":"true","extra":"keeper"}, 
 {"keep":"false","extra":"non-keeper"}]

此时很容易过滤所需的对象。但是，这会从其父对象的上下文中剥离结果，这不是我想要的。

查看流式结构：

[["keep_untouched","keep_this",0],"this"]
[["keep_untouched","keep_this",1],"list"]
[["keep_untouched","keep_this",1]]
[["keep_untouched","keep_this"]]
[["filter_this",0,"keep"],"true"]
[["filter_this",0,"keep"]]
[["filter_this",1,"keep"],"true"]
[["filter_this",1,"extra"],"keeper"]
[["filter_this",1,"extra"]]
[["filter_this",2,"keep"],"false"] …