flink-user mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From "Nortman, Bill" <William.Nort...@pimco.com>
Subject CEP Debugging
Date Thu, 19 Apr 2018 21:51:25 GMT
So I have events coming in in this format
TransactionID, MicroserviceName, Direction,EpochTimeStamp.
For each call to a microservice and event is generates with a timestamp with direction of
"in". Then when completed it generates with a timestamp with direction of "out".
I need to calculate latency for both the microservices per transactions and for the entire
Example Data (TransactionID, MicroserviceName, Direction,EpochTimeStamp)
Look for a pattern to handle this.
Something like this
        Pattern<MetricsEvent,?> pattern = Pattern.<MetricsEvent>begin("myBegin").where(
            new SimpleCondition<MetricsEvent>() {
                public boolean filter(MetricsEvent metricsEvent) throws Exception {
                    return metricsEvent.getDirection().equals("in");
            new SimpleCondition<MetricsEvent>() {
                public boolean filter(MetricsEvent metricsEvent) throws Exception {
                    metricsEvent.getApplication().equals(//PREVIOUS EVENT.getApplication()))
                    return false;
I'm note sure how to get the previous event to compare too.
Then how to calculate the latency between the two events?
So I tried this
    Pattern<MetricsEvent, ?> pattern = Pattern.<MetricsEvent>begin("myBegin").where(
            new SimpleCondition<MetricsEvent>() {
                public boolean filter(final MetricsEvent metricsEvent)  {
                    System.out.println("Begin Filter");
                    return !metricsEvent.getEventType().equals("in");

            new SimpleCondition<MetricsEvent>() {
                public boolean filter(final MetricsEvent metricsEvent) {
                    return !metricsEvent.getEventType().equals("out");
    // Define a stream processor using the pattern
    PatternStream<MetricsEvent> patternStream = CEP.pattern(

    // Process the stream
    System.out.println("************* Pattern Processing");
    SingleOutputStreamOperator<MetricLatency> latencyEvents =
            patternStream.flatSelect(new ApplicationLatencyFlatSelectFunction());
With this as the latency calc
public class ApplicationLatencyFlatSelectFunction implements
    org.apache.flink.cep.PatternFlatSelectFunction<MetricsEvent, MetricLatency> {
@SuppressWarnings("CheckStyle") // Base class method doesn't have JavaDoc
public void flatSelect(final Map<String, List<MetricsEvent>> map, final Collector<MetricLatency>
collector) {
        System.out.println("zzzzzzzzzzzzzzzzzzzz  Flat Select From Pattern");
        MetricsEvent begin = map.get("myBegin").get(0);
        List<MetricsEvent> ends = map.get("followed");
        for (MetricsEvent me: ends
                ) {
            if (me.getApplication().equals(begin.getApplication())) {
                Long latency = me.getEpochTime() - begin.getEpochTime();
                collector.collect(new MetricLatency(begin.getUid(), begin.getApplication(),

However I don't get any output, the printlines in the pattern or the flatselect function never
How do you debug something like this?

This message contains confidential information and is intended only for the individual named.
If you are not the named addressee, you should not disseminate, distribute, alter or copy
this e-mail. Please notify the sender immediately by e-mail if you have received this e-mail
by mistake and delete this e-mail from your system. E-mail transmissions cannot be guaranteed
to be secure or without error as information could be intercepted, corrupted, lost, destroyed,
arrive late or incomplete, or contain viruses. The sender, therefore, does not accept liability
for any errors or omissions in the contents of this message which arise during or as a result
of e-mail transmission. If verification is required, please request a hard-copy version. This
message is provided for information purposes and should not be construed as a solicitation
or offer to buy or sell any securities or related financial instruments in any jurisdiction.
 Securities are offered in the U.S. through PIMCO Investments LLC, distributor and a company

The individual providing the information herein is an employee of Pacific Investment Management
Company LLC ("PIMCO"), an SEC-registered investment adviser.  To the extent such individual
advises you regarding a PIMCO investment strategy, he or she does so as an associated person
of PIMCO.  To the extent that any information is provided to you related to a PIMCO-sponsored
investment fund ("PIMCO Fund"), it is being provided to you in the individual's capacity as
a registered representative of PIMCO Investments LLC ("PI"), an SEC-registered broker-dealer.
 PI is not registered, and does not intend to register, as a municipal advisor and therefore
does not provide advice with respect to the investment of the proceeds of municipal securities
or municipal escrow investments.  In addition, unless otherwise agreed by PIMCO, this communication
and any related attachments are being provided on the express basis that they will not cause
PIMCO LLC, or its affiliates, to become an investment advice fiduciary under ERISA or the
Internal Revenue Code.

View raw message