Mailing-List: contact user-help@flink.apache.org; run by ezmlm
Precedence: bulk
Reply-To: user@flink.apache.org
From: "kieran ." <kieran0628@hotmail.com>
To: "user@flink.apache.org" <user@flink.apache.org>
Subject: CEP issue
Thread-Topic: CEP issue
Thread-Index: AQHSSksLMOtVC5C3FUWwTKpzC2jx5Q==
Date: Tue, 29 Nov 2016 14:18:40 +0000
Message-ID: <DB5PR06MB1829C6FFACFA3AF040AE048AD98D0@DB5PR06MB1829.eurprd06.prod.outlook.com>
Accept-Language: en-GB, en-US
Content-Language: en-GB
spamdiagnosticoutput: 1:99
spamdiagnosticmetadata: NSPM
Content-Type: multipart/alternative;
	boundary="_000_DB5PR06MB1829C6FFACFA3AF040AE048AD98D0DB5PR06MB1829eurp_"
MIME-Version: 1.0
X-MS-Exchange-CrossTenant-originalarrivaltime: 29 Nov 2016 14:18:40.4682
 (UTC)
X-MS-Exchange-CrossTenant-fromentityheader: Internet
X-MS-Exchange-CrossTenant-id: 84df9e7f-e9f6-40af-b435-aaaaaaaaaaaa
X-MS-Exchange-Transport-CrossTenantHeadersStamped: DB5EUR03HT184
X-OriginalArrivalTime: 29 Nov 2016 14:18:42.0248 (UTC) FILETIME=[7CB62480:01D24A4B]
archived-at: Tue, 29 Nov 2016 14:19:36 -0000

--_000_DB5PR06MB1829C6FFACFA3AF040AE048AD98D0DB5PR06MB1829eurp_
Content-Type: text/plain; charset="iso-8859-1"
Content-Transfer-Encoding: quoted-printable

Hello,

I am currently building a multi-tenant monitoring application and exploring=
 the effectiveness of different Complex Event Processors (CEP) and whether =
or not this would be a potential solution for what I want to achieve. I hav=
e created a small test application which utilises Flink and its CEP but I h=
ave come across some issues when dealing with a large number of metrics to =
monitor when using patterns/pattern streams. Flink seems to operate as expe=
cted with one, or several patterns each consuming it's own PatternStream, b=
ut as soon as more are introduced the memory usage of Flink seems to rise r=
ather quickly and eventually throw an OutOfMemoryError. My initial idea was=
 to create one pattern/pattern stream for each metric that I need to monito=
r, but there could be many thousands of these.

I create the PatternStream per Pattern like this to monitor a metric:


  Pattern<MetricData, ?> pattern =3D Pattern.<MetricData> begin( patternNam=
e ).subtype( MetricData.class )

                .where(

                (evt -> evt.getValues().get( "max" ).longValue() > 50.0

                        && evt.account_id.equals( accountName )) );


        check.withPattern( pattern )

                .withTimePeriod( Integer.valueOf( 1 ) )

                .withCooldown( Integer.valueOf( 1 ) )

                .withName( checkName )

                .withAlertStatus( AlertStatus.OK )

                .setPatternStream(CEP.pattern(messageStream.keyBy("account_=
id"), pattern));

To trigger these patterns, I use


        PatternSelectFunction<MetricData, MetricWarning> psf =3D new Patter=
nSelectFunction<MetricData, MetricWarning>()

        {

            @Override

            public MetricWarning select( Map<String, MetricData> map ) thro=
ws Exception

            {

                return new MetricWarning(map.get(patternKey), name, account=
Id);

            }


        };


        try

        {

            check.getPatternStream().select(psf);

        }

        catch( Exception exception )

        {

            exception.printStackTrace();

        }


The pattern in the above example is tied to a specific stream which would r=
esult in one stream per pattern and this seems to be an issue using this ap=
proach. If it would be possible to run one pattern stream and switching out=
 the patterns when needed, then perhaps this would be a viable solution. Am=
 I approaching this in the right way by creating a stream for each pattern?

Would it be possible to create a set of Pattern processors that could be ru=
n against a single PatternStream or is there anything you could suggest whi=
ch would allow me to do this with Flink?

Thanks,
- Kieran

<http://aka.ms/weboutlook>

--_000_DB5PR06MB1829C6FFACFA3AF040AE048AD98D0DB5PR06MB1829eurp_
Content-Type: text/html; charset="iso-8859-1"
Content-Transfer-Encoding: quoted-printable

<html>
<head>
<meta http-equiv=3D"Content-Type" content=3D"text/html; charset=3Diso-8859-=
1">
<style type=3D"text/css" style=3D"display:none;"><!-- P {margin-top:0;margi=
n-bottom:0;} --></style>
</head>
<body dir=3D"ltr">
<div id=3D"divtagdefaultwrapper" style=3D"font-size:12pt;color:#000000;font=
-family:Calibri,Arial,Helvetica,sans-serif;" dir=3D"ltr">
<p></p>
<div>Hello,</div>
<div><br>
</div>
<div>I am currently building a multi-tenant monitoring application and expl=
oring the effectiveness of different Complex Event Processors (CEP) and whe=
ther or not this would be a potential solution for what I want to achieve.&=
nbsp;<span style=3D"font-size: 12pt;">I
 have created a small test application which utilises Flink and its CEP but=
 I have come across some issues when dealing with a large number of metrics=
 to monitor when using patterns/pattern streams.
</span><span style=3D"font-size: 12pt;">Flink seems to operate as expected =
with one, or several patterns each consuming it's own PatternStream, but as=
 soon as more are introduced the memory usage of&nbsp;</span><span style=3D=
"font-size: 12pt;">Flink seems to rise rather
 quickly and eventually throw an OutOfMemoryError. My initial idea was to c=
reate one pattern/pattern stream for each metric that I need to monitor, bu=
t there could be many thousands of these.</span></div>
<div><br>
</div>
<div>I create the PatternStream per Pattern like this to monitor a metric:<=
/div>
<div><br>
</div>
<p></p>
<blockquote style=3D"margin: 0 0 0 40px; border: none; padding: 0px;">
<p></p>
<div><i>&nbsp;<span class=3D"Apple-tab-span" style=3D"white-space:pre"> </s=
pan>Pattern&lt;MetricData, ?&gt; pattern =3D Pattern.&lt;MetricData&gt; beg=
in( patternName ).subtype( MetricData.class )</i></div>
<p></p>
<p></p>
<div><i>&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; .where(</i>=
</div>
<p></p>
<p></p>
<div><i>&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; (evt -&gt; =
evt.getValues().get( &quot;max&quot; ).longValue() &gt; 50.0</i></div>
<p></p>
<p></p>
<div><i>&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbs=
p; &nbsp; &nbsp; &amp;&amp; evt.account_id.equals( accountName )) );</i></d=
iv>
<p></p>
<p></p>
<div><i><br>
</i></div>
<p></p>
<p></p>
<div><i>&nbsp; &nbsp; &nbsp; &nbsp; check.withPattern( pattern )</i></div>
<p></p>
<p></p>
<div><i>&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; .withTimePe=
riod( Integer.valueOf( 1 ) )</i></div>
<p></p>
<p></p>
<div><i>&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; .withCooldo=
wn( Integer.valueOf( 1 ) )</i></div>
<p></p>
<p></p>
<div><i>&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; .withName( =
checkName )</i></div>
<p></p>
<p></p>
<div><i>&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; .withAlertS=
tatus( AlertStatus.OK )</i></div>
<p></p>
<p></p>
<div><i>&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; .setPattern=
Stream(CEP.pattern(messageStream.keyBy(&quot;account_id&quot;), pattern));<=
/i></div>
<p></p>
</blockquote>
<p></p>
<div><br>
</div>
<div>To trigger these patterns, I use</div>
<div><br>
</div>
<p></p>
<blockquote style=3D"margin: 0 0 0 40px; border: none; padding: 0px;">
<p></p>
<div><i>&nbsp; &nbsp; &nbsp; &nbsp; PatternSelectFunction&lt;MetricData, Me=
tricWarning&gt; psf =3D new PatternSelectFunction&lt;MetricData, MetricWarn=
ing&gt;()</i></div>
<p></p>
<p></p>
<div><i>&nbsp; &nbsp; &nbsp; &nbsp; {</i></div>
<p></p>
<p></p>
<div><i>&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; @Override</i></div>
<p></p>
<p></p>
<div><i>&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; public MetricWarning sele=
ct( Map&lt;String, MetricData&gt; map ) throws Exception</i></div>
<p></p>
<p></p>
<div><i>&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; {</i></div>
<p></p>
<p></p>
<div><i>&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; return new =
MetricWarning(map.get(patternKey), name, accountId);</i></div>
<p></p>
<p></p>
<div><i>&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; }</i></div>
<p></p>
<p></p>
<div><i><br>
</i></div>
<p></p>
<p></p>
<div><i>&nbsp; &nbsp; &nbsp; &nbsp; };</i></div>
<p></p>
<p></p>
<div><i><br>
</i></div>
<p></p>
<p></p>
<div><i>&nbsp; &nbsp; &nbsp; &nbsp; try</i></div>
<p></p>
<p></p>
<div><i>&nbsp; &nbsp; &nbsp; &nbsp; {</i></div>
<p></p>
<p></p>
<div><i>&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; check.getPatternStream().=
select(psf);</i></div>
<p></p>
<p></p>
<div><i>&nbsp; &nbsp; &nbsp; &nbsp; }</i></div>
<p></p>
<p></p>
<div><i>&nbsp; &nbsp; &nbsp; &nbsp; catch( Exception exception )</i></div>
<p></p>
<p></p>
<div><i>&nbsp; &nbsp; &nbsp; &nbsp; {</i></div>
<p></p>
<p></p>
<div><i>&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; exception.printStackTrace=
();</i></div>
<p></p>
<p></p>
<div><i>&nbsp; &nbsp; &nbsp; &nbsp; }</i></div>
<p></p>
</blockquote>
<p></p>
<div><br>
</div>
<div><br>
</div>
<div>The pattern in the above example is tied to a specific stream which wo=
uld result in one stream per pattern and this seems to be an issue using th=
is approach. If it would be possible to run one pattern stream and switchin=
g out the patterns when needed,
 then perhaps this would be a viable solution. Am I approaching this in the=
 right way by creating a stream for each pattern?&nbsp;</div>
<div><br>
</div>
<div>Would it be possible to create a set of Pattern processors that could =
be run against a single PatternStream or is there anything you could sugges=
t which would allow me to do this with Flink?</div>
<div><br>
</div>
<div>Thanks,</div>
<div>- Kieran</div>
<p></p>
<div id=3D"Signature">
<div id=3D"divtagdefaultwrapper" style=3D"font-size:12pt; color:#000000; ba=
ckground-color:#FFFFFF; font-family:Calibri,Arial,Helvetica,sans-serif">
<a href=3D"http://aka.ms/weboutlook" id=3D"LPNoLP"></a></div>
</div>
</div>
</body>
</html>

--_000_DB5PR06MB1829C6FFACFA3AF040AE048AD98D0DB5PR06MB1829eurp_--