标签: dataflow

例如,我会写:

x = 2
y = x + 4
print(y)
x = 5
print(y)

它会输出:

6 (=2+4)
9 (=5+4)

此外,是否有任何可能实际有用的情况？

澄清:是的,lambdas等解决了这个问题(他们是我如何得出这个想法); 我想知道是否存在默认的特定语言:不需要或不需要函数或lambda关键字.

我想绘制一个图表,您可以在其中查看java程序的数据流,以及是否有一个或多个线程处理数据.

当您有超过5个不同的实体时,序列图不会显示多线程并且会非常混乱.

类图适用于显示关系,但缺少数据流.

我想绘制框(如类/包图)并连接它们.也许有一个盒子加倍,...介于两者之间,以显示可以有多个线程.

UML对这种数据流图有什么图表？

什么编辑器对该图有好处？如果可能的免费软件或开源.

之前曾问过这个问题,但答案非常通用.我很想知道在静态代码分析的上下文中有什么上下文,特别是Java,以及与术语context(in)敏感分析结合使用时.

例如,本文在此背景下广泛使用"背景".事实上,我还没有找到合适的上下文定义.

如何描述"数据流编程"和"演员模型"之间的区别？据我所知,它们并非无关,但却不一样.DF是一个更广泛的概念,其主旨是与控制流模型的区别,而Actor模型是DF的更精细和理论基础的继承者？

我试图在 Verilog 中计算 4 位二进制数中 1 的数量，但我的输出出乎意料。我尝试了几种方法；这是我认为应该工作的一个，但它没有。

module ones(one,in);
input [3:0]in;
output [1:0]one;

assign one = 2'b00; 
assign one = one+in[3]+in[2]+in[1]+in[0] ;

endmodule

我注意到java apache beam有类groupby.sortbytimestamp python是否已实现该功能？如果不是在窗口中对元素进行排序的方法是什么？我想我可以在DoFn中对整个窗口进行排序,但我想知道是否有更好的方法.

考虑以下示例：

        ActionBlock<TimeSpan> ab = new ActionBlock<TimeSpan>(async _ =>
        {
            await Task.Delay(_);
            throw new Exception();
        }, new ExecutionDataflowBlockOptions() { MaxDegreeOfParallelism = Int32.MaxValue });

        ab.Post(TimeSpan.FromSeconds(10d));
        ab.Post(TimeSpan.FromDays(1d));

        await ab.Completion;

不出所料，我在输出窗口中看到10秒后将引发异常，但这不会导致dataFlow完成（等待ab之后的断点。完成将不会在1天前命中）。

就我而言，如果单个计算步骤中有异常，我想取消整个dataFlow。

我看不到如何使用tpl dataflow做到这一点...有什么建议吗？

谢谢...

[edit]如Ofir所述，我可以这样做：

        ActionBlock<TimeSpan> ab = new ActionBlock<TimeSpan>(async _ =>
        {
            try
            {
                await Task.Delay(_);
                throw new Exception();// Or any other Task thay may throw an exception.
            }
            catch(Exception)
            {
                cancelTokenSource.Cancel();
                throw;
            }
        }, new ExecutionDataflowBlockOptions() {CancellationToken=cancelTokenSource.Token, MaxDegreeOfParallelism = Int32.MaxValue });

        ab.Post(TimeSpan.FromSeconds(10d));
        ab.Post(TimeSpan.FromDays(1d));

        await ab.Completion;

但这恰恰是我要避免的事情：一次又一次地键入相同的try {} catch……或更糟糕的是：忘记了...;）

我接下来要做的是写一个替换ActionBlock的构造函数来处理该问题的方法（这将需要一个额外的CancellationTokenSource参数）...我很惊讶这不是直接与数据流一起出现的吗？确实如此吗？

[最终编辑]似乎答案是tpd …

我正在尝试使用数据流读取 pubsub 消息并将其写入大查询。我获得了 Google 团队的 alpha 访问权限，并且已经使提供的示例正常工作，但现在我需要将其应用到我的场景中。

发布订阅负载：

Message {
    data: {'datetime': '2017-07-13T21:15:02Z', 'mac': 'FC:FC:48:AE:F6:94', 'status': 1}
    attributes: {}
}

大查询架构：

schema='mac:STRING, status:INTEGER, datetime:TIMESTAMP',

我的目标是简单地读取消息有效负载并插入到 bigquery 中。我正在努力了解转换以及如何将键/值映射到大查询模式。

我对此很陌生，因此非常感谢任何帮助。

当前代码： https: //codeshare.io/ayqX8w

谢谢！

我需要制作可扩展的流程.该进程主要具有I/O操作和一些次要CPU操作(主要是反序列化字符串).该进程在数据库中查询url列表,然后从这些url中获取数据,将下载的数据deserilize到对象,然后将一些数据保存到crm动态以及另一个数据库中.之后我需要更新第一个处理网址的数据库.部分要求是使并行度可配置.

最初我想通过一系列任务实现它,等待并使用Semaphore限制并行性 - 非常简单.然后我读了@Stephen Cleary的一些帖子和答案,建议使用TPL Dataflow,我认为它可能是一个很好的候选人.但是,我想通过使用Dataflow来确保我使代码"复杂化",这是值得的.我也有一个建议使用ForEachAsync扩展方法也很简单,但是我不确定它是否会因为它对集合进行分区而导致内存开销.

TPL Dataflow是否适合这种情况？它如何比Semaphore或ForEachAsync方法更好 - 如果我通过TPL DataFlow在每个其他选项(Semaphore/ForEachASync)上实现它,我将获得什么好处？

我需要为我正在处理的用例之一提供一些建议。

用例：

我们在 Cloud SQL 中有大约 5-10 个表的数据，有些被视为查找，有些被视为事务性的。我们需要以某种方式将其提供给 BigQuery，从中制作 3-4 个表（扁平化、嵌套或非规范化），这些表将用于 Data Studio、Looker 等中的报告。

数据应该以增量方式处理，并且 Cloud SQL 中的更改可能每 5 分钟发生一次，这意味着数据应该可以近乎实时地用于 BigQuery 报告。如果可能，我们还应该维护数据更改的历史记录。

解决方案：

我们在 BigQuery 中保留了 3 层，因此来自 Cloud SQL 的数据来到第一层，然后在展平后将其保存到第二层（维护历史记录），然后将其处理插入第三层（报告）。这种方法的 问题在于，它是为日常处理而开发的，无法帮助处理 Cloud SQL 中每 5 分钟更改一次的数据。

欢迎任何建议。

注意：-我们需要使用 Dataflow、VM 或其他方式来进行一些逻辑处理，这意味着我们不能只使用计划查询，因为我们不能使用查询来处理所有逻辑，但是可以使用视图和代码的混合.