Gibt es eine Möglichkeit, Tupel/Sek durch einen Operator in IBM Streams (nicht durch Streams-Konsole) zu cappen

Question

Ich möchte die Anzahl der Tupel/Sek durch einen Operator erfassen und in einer Datei protokollieren. Ich kann 'Throttle Operator' nicht verwenden, um die Tupelrate selbst einzustellen. Um noch einmal hinzuzufügen, ich spreche nicht über die Erfassung der Informationen über die Konsole, sondern durch SPL-Anwendung.

Answer 1

Es gibt keine direkte "Geben Sie mir den Durchsatz für diesen Operator" Metrik verfügbar. Sie könnten einen primitiven Operator implementieren, der im Zeitverlauf auf die nTuplesProcessed Metrik zugreift und daraus den Durchsatz berechnet. (Die list of available metrics.) Aber, ich habe es tatsächlich finden viel einfacher, den folgenden Verbund Operator verwenden:

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; 
    } 
} 

Anschließend können Sie den Composite-Operator als „Durchsatz tap“ verwenden, wo es den Strom, dessen von welchem ​​Betreiber verbraucht Durchsatz, den Sie aufzeichnen möchten. Zum Beispiel können Sie es wie so verwenden:

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

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

Natürlich können Sie dann noch den Result Strom an anderer Stelle in Ihrer Anwendung verwenden. Beachten Sie, dass diese Methode möglicherweise Auswirkungen auf die Leistung der Anwendung hat: Wir tippen auf den Datenpfad. Die Auswirkungen sollten jedoch nicht groß sein, insbesondere, wenn Sie sicherstellen, dass die Operatoren in PeriodicThroughputSink mit demselben Operator verknüpft sind wie jeder Operator, auf den Sie tippen. Je kürzer der Zeitraum ist, desto wahrscheinlicher wird es die Anwendungsleistung beeinflussen.

Wieder könnten wir etwas Ähnliches in einem C++ - oder Java-Primitivoperator tun, indem wir auf die nTuplesProcessed Metrik zugreifen, aber ich finde den obigen Ansatz viel einfacher. Sie können die Systemmetriken auch außerhalb Ihrer Anwendung abrufen. Sagen wir, Sie könnten ein Skript haben, das in regelmäßigen Abständen streamtool capturestate oder die REST-API verwendet, und dann die Ausgabe analysieren, die Metrik nTuplesProcessed für den Operator finden, den Sie interessieren, und diesen zur Berechnung des Durchsatzes verwenden. Aber ich finde die Technik in diesem zusammengesetzten Operator viel einfacher.