0

我想通过运算符捕获元组/秒的数量并将其记录在文件中。我不能使用“Throttle Operator”自己设置元组速率。另外，再次补充一下，我不是在谈论通过控制台捕获信息，而是通过 SPL 应用程序。

4

1 回答 1

2

没有可用的直接“给我这个操作员的吞吐量”指标。您可以实现一个原始运算符，该运算符nTuplesProcessed随时间访问指标并从中计算吞吐量。（可用指标列表。）但是，我实际上发现使用以下复合运算符要容易得多：

public composite PeriodicThroughputSink(input In) {
param expression<float64> $period;
      expression<rstring> $file;
graph
    stream<boolean b> Period = Beacon() {
        param period: $period;
    }

    stream<float64 throughput> Throughput = Custom(In; Period) {
        logic state: {
            mutable uint64 _count = 0;
            float64 _period = $period;
        }

        onTuple In: {
            ++_count;
        }

        onTuple Period: {
            if (_count > 0ul) {
                submit({throughput=((float64)_count / _period)}, Throughput);
                _count = 0ul;
            }
        }

        config threadedPort: queue(Period, Sys.Wait); // ensures that the throughput calculation and file
                                                      // writing is on a different thread from the rest 
                                                      // of the application
    }

    () as Sink = FileSink(Throughput) {
        param file: $file;
              format: txt;
              flush: 1u;
    }
}

然后，您可以将复合运算符用作“吞吐量抽头”，它使用来自您要记录其吞吐量的任何运算符的流。例如，您可以像这样使用它：

stream<Data> Result = OperatorYouCareAbout(In) {}

() as ResultThroughput = PeriodicThroughputSink(Result) {
    param period: 5.0;
          file: "ResultThroughput.txt";
}

当然，您仍然可以Result在应用程序的其他地方使用流。请记住，这种方法可能会对应用程序的性能产生一些影响：我们正在点击数据路径。但是，影响不应该很大，特别是如果您确保其中的运算符PeriodicThroughputSink与您正在挖掘的任何运算符融合到同一个 PE 中。此外，时间越短，影响应用程序性能的可能性就越大。

同样，我们可以通过访问度量在 C++ 或 Java 原始运算符中做类似的事情nTuplesProcessed，但我发现上述方法要容易得多。您还可以从应用程序外部获取系统指标；比如说，您可以有一个定期使用的脚本streamtool capturestate或 REST API，然后解析输出，找到nTuplesProcessed您关心的运营商的指标并使用它来计算吞吐量。但我发现这个复合运算符中的技术要容易得多。

于 2017-08-24T13:59:07.250 回答