在级联中创建“悬挂”管道答案

【问题标题】：Creating a "hanging" pipe in cascading在级联中创建“悬挂”管道
【发布时间】：2015-05-22 14:17:29
【问题描述】：

基本上在我的 FlowDef 中，我想将我的元组与我将通过 API 读取的其他一些元组连接起来。所以我创建了另一个空管道，并使用函数从 API 中填充元组。然后我做一个 CoGroup 并将其用作尾管。

Tap tap1 = new Hfs();
Pipe pipe1 = new Pipe("pipe1");

Pipe pipeFromApi = new Pipe("pipeFromApi");
Pipe joinPipe = new CoGroup(pipe1, pipeFromApi);
Tap tap2 = new Hfs();

flowDef.addSource(pipe1, tap1).addTailSink(joinPipe, tap2);

这会导致以下错误 - “在接收器或源映射中找不到管道名称：'pipeFromApi'”。

关于如何实现这一点的任何想法？

【问题讨论】：

标签： cascading

【解决方案1】：

您可以创建一个虚拟 SinkTap 并将其设为 pipeFromApi 的“源”。

例外情况是管道“pipeFromApi”没有水龙头 - 而每个管道都应该有一个。

【讨论】：

【解决方案2】：

Cascading 中的 Pipe 用于命名管道组件的分支，在规划期间将 Taps 绑定为源或汇时使用。 Tap 是 Pipe 组件的数据源或数据接收器。从这些定义中可以很清楚地看出，为了承载来自 Tap 的数据，您需要将其与 Pipe 绑定。在您的代码中，您已经定义了 Pipe pipeFromApi 但尚未为此创建任何源（Tap）。此外，创建此 Tap 后，您需要将 Pipe pipeFromApi 与 flowDef 中新创建的 Tap 绑定。然后只有流能够识别两个输入源，一个是您的 pipe1，另一个是您的 pipeFromApi。所以正确的代码 sn-p 可能是这样的：

Tap tap1 = new Hfs("..inputSource");
Pipe pipe1 = new Pipe("pipe1");

Tap tap2 = new Hfs("..inputSourceFromApi");
Pipe pipeFromApi = new Pipe("pipeFromApi");

Tap tap3 = new Hfs("..outputSinkDestination");
Pipe joinPipe = new CoGroup(pipe1, pipeFromApi);

flowDef.addSource(pipe1, tap1).addSource(pipeFromApi, tap2).addTailSink(joinPipe, tap3);

【讨论】：