DAG源在emitFromTraverser上返回false，处理器在开始处理之前等待源加载的所有元素

Question

给定DAG的这个（simpized版本）

发出5个A类型项的S1源（从带有分区的数据库中读取）本地并行度=1

S2发出150k个B类型项的源代码（迭代器，从DB中批量读取100个，并进行分区）本地并行度=1

写入B1本地并行度=1的WR接收器

注意：为了给过滤器处理器赋予意义：在DAG中，有其他源流入同一个适配器AD，然后使用过滤器处理器进入其他路径。

S1-->AD

FB-->CL（到序数1，优先级为1）

CL-->WR

如果源S2有“很少”项要加载（即15K)，则emitFromTraverser从不返回false。

null

protected void init(Context context) throws Exception {
    super.init(context);
    this.iterator = new BQueryIterator(querySupplier, batchSize);
    this.traverser = Traversers.traverseIterator(this.iterator);
}

public boolean complete() {
    boolean result = emitFromTraverser(this.traverser);
    return result;
}

null

谢了！

Answer 1

您遭受由优先级导致的死锁。您的DAG从AD分支，然后重新加入CL，但有一个优先级。

AD --+---- FA ----+-- CL
      \          /
       +-- FB --+

设置优先级会导致在处理来自高优先级边缘的所有项之前，不处理来自低优先级边缘的项。AD最终会被来自低优先级路径的反压阻塞，而CL不会处理该路径。因此，ad被阻塞，因为它不能发射到低优先级边缘，而CL被阻塞，因为它仍然在等待来自高优先级边缘的项，从而导致死锁。

在您的例子中，您可以通过创建2个ad顶点来解决这个问题，每个顶点处理来自其中一个源的项：

S1 --- AD1 ----+--- CL
              /
S2 --- AD2 --+