Mailing-List: contact user-help@flink.apache.org; run by ezmlm
Precedence: bulk
MIME-Version: 1.0
In-Reply-To: <A5BC0225-C295-422B-BB28-B1914F27D1EE@data-artisans.com>
References: <CANwfUVN-6NwX4tROx+fq8ASP8Y39NEaWs0qN3Pd7JEcFGvkr6A@mail.gmail.com>
 <A5BC0225-C295-422B-BB28-B1914F27D1EE@data-artisans.com>
From: Shailesh Jain <shailesh.jain@stellapps.com>
Date: Thu, 9 Nov 2017 23:46:26 +0530
Message-ID: <CANwfUVNp2MmDanjBQgHJgTyifdYGWtU3d0VrV6GULjjDeuaWnw@mail.gmail.com>
Subject: Re: Correlation between data streams/operators and threads
To: Piotr Nowojski <piotr@data-artisans.com>
Cc: user@flink.apache.org
Content-Type: multipart/alternative; boundary="94eb2c082cbe7c72a0055d90cd29"
archived-at: Thu, 09 Nov 2017 18:16:33 -0000

--94eb2c082cbe7c72a0055d90cd29
Content-Type: text/plain; charset="UTF-8"
Content-Transfer-Encoding: quoted-printable

On 1. - is it tied specifically to the number of source operators or to the
number of Datastream objects created. I mean does the answer change if I
read all the data from a single Kafka topic, get a Datastream of all
events, and the apply N filters to create N individual streams?

On 3. - the problem with partitions is that watermarks cannot be different
per partition, and since in this use case, each stream is from a device,
the latency could be different (but order will be correct almost always)
and there are high chances of loosing out on events on operators like
Patterns which work with windows. Any ideas for workarounds here?


Thanks,
Shailesh

On 09-Nov-2017 8:48 PM, "Piotr Nowojski" <piotr@data-artisans.com> wrote:

Hi,

1.
https://ci.apache.org/projects/flink/flink-docs-
release-1.3/dev/parallel.html

Number of threads executing would be roughly speaking equal to of the
number of input data streams multiplied by the parallelism.

2.
Yes, you could dynamically create more data streams at the job startup.

3.
Running 10000 independent data streams on a small cluster (couple of nodes)
will definitely be an issue, since even with parallelism set to 1, there
would be quite a lot of unnecessary threads.

It would be much better to treat your data as a single data input stream
with multiple partitions. You could assign partitions between source
instances based on parallelism. For example with parallelism 6:
- source 0 could get partitions 0, 6, 12, 18
- source 1, could get partitions 1, 7, =E2=80=A6
=E2=80=A6
- source 5, could get partitions 5, 11, ...

Piotrek

On 9 Nov 2017, at 10:18, Shailesh Jain <shailesh.jain@stellapps.com> wrote:

Hi,

I'm trying to understand the runtime aspect of Flink when dealing with
multiple data streams and multiple operators per data stream.

Use case: N data streams in a single flink job (each data stream
representing 1 device - with different time latencies), and each of these
data streams gets split into two streams, of which one goes into a bunch of
CEP operators, and one into a process function.

Questions:
1. At runtime, will the engine create one thread per data stream? Or one
thread per operator?
2. Is it possible to dynamically create a data stream at runtime when the
job starts? (i.e. if N is read from a file when the job starts and
corresponding N streams need to be created)
3. Are there any specific performance impacts when a large number of
streams (N ~ 10000) are created, as opposed to N partitions within a single
stream?

Are there any internal (design) documents which can help understanding the
implementation details? Any references to the source will also be really
helpful.

Thanks in advance.

Shailesh

--94eb2c082cbe7c72a0055d90cd29
Content-Type: text/html; charset="UTF-8"
Content-Transfer-Encoding: quoted-printable

<div dir=3D"auto"><div>On 1. - is it tied specifically to the number of sou=
rce operators or to the number of Datastream objects created. I mean does t=
he answer change if I read all the data from a single Kafka topic, get a Da=
tastream of all events, and the apply N filters to create N individual stre=
ams?</div><div dir=3D"auto"><br></div><div dir=3D"auto">On 3. - the problem=
 with partitions is that watermarks cannot be different per partition, and =
since in this use case, each stream is from a device, the latency could be =
different (but order will be correct almost always) and there are high chan=
ces of loosing out on events on operators like Patterns which work with win=
dows. Any ideas for workarounds here?</div><div dir=3D"auto"><div class=3D"=
gmail_extra" dir=3D"auto"><br></div><div class=3D"gmail_extra" dir=3D"auto"=
><br></div><div class=3D"gmail_extra" dir=3D"auto">Thanks,</div><div class=
=3D"gmail_extra" dir=3D"auto">Shailesh</div><div class=3D"gmail_extra" dir=
=3D"auto"><br><div class=3D"gmail_quote" dir=3D"auto">On 09-Nov-2017 8:48 P=
M, &quot;Piotr Nowojski&quot; &lt;<a href=3D"mailto:piotr@data-artisans.com=
">piotr@data-artisans.com</a>&gt; wrote:<br type=3D"attribution"><blockquot=
e class=3D"quote" style=3D"margin:0 0 0 .8ex;border-left:1px #ccc solid;pad=
ding-left:1ex"><div style=3D"word-wrap:break-word">Hi,<div><br></div><div>1=
.=C2=A0</div><div><a href=3D"https://ci.apache.org/projects/flink/flink-doc=
s-release-1.3/dev/parallel.html" target=3D"_blank">https://ci.apache.org/<w=
br>projects/flink/flink-docs-<wbr>release-1.3/dev/parallel.html</a></div><d=
iv><br></div><div>Number of threads executing would be roughly speaking equ=
al to of the number of input data streams multiplied by the parallelism.</d=
iv><div><br></div><div>2.=C2=A0</div><div>Yes, you could dynamically create=
 more data streams at the job startup.</div><div><br></div><div>3.</div><di=
v>Running 10000 independent data streams on a small cluster (couple of node=
s) will definitely be an issue, since even with parallelism set to 1, there=
 would be quite a lot of unnecessary threads.=C2=A0</div><div><br></div><di=
v>It would be much better to treat your data as a single data input stream =
with multiple partitions. You could assign partitions between source instan=
ces based on parallelism. For example with parallelism 6:</div><div>- sourc=
e 0 could get partitions 0, 6, 12, 18</div><div>- source 1, could get parti=
tions 1, 7, =E2=80=A6</div><div>=E2=80=A6</div><div>- source 5, could get p=
artitions 5, 11, ...</div><div><br></div><div>Piotrek</div><div class=3D"el=
ided-text"><div><br><div><blockquote type=3D"cite"><div>On 9 Nov 2017, at 1=
0:18, Shailesh Jain &lt;<a href=3D"mailto:shailesh.jain@stellapps.com" targ=
et=3D"_blank">shailesh.jain@stellapps.com</a>&gt; wrote:</div><br class=3D"=
m_1755883704455667468Apple-interchange-newline"><div><div dir=3D"ltr"><div>=
<div><div><div><div><div><div><div><div><div>Hi,<br><br></div>I&#39;m tryin=
g to understand the runtime aspect of Flink when dealing with multiple data=
 streams and multiple operators per data stream.<br></div><div><br></div>Us=
e case: N data streams in a single flink job (each data stream representing=
 1 device - with different time latencies), and each of these data streams =
gets split into two streams, of which one goes into a bunch of CEP operator=
s, and one into a process function.<br><br></div>Questions:<br></div>1. At =
runtime, will the engine create one thread per data stream? Or one thread p=
er operator?<br></div></div>2. Is it possible to dynamically create a data =
stream at runtime when the job starts? (i.e. if N is read from a file when =
the job starts and corresponding N streams need to be created)<br></div>3. =
Are there any specific performance impacts when a large number of streams (=
N ~ 10000) are created, as opposed to N partitions within a single stream?<=
br><br></div>Are there any internal (design) documents which can help under=
standing the implementation details? Any references to the source will also=
 be really helpful.<br><br></div>Thanks in advance.<br><br></div>Shailesh<b=
r><div><div><div><div><br><br></div></div></div></div></div>
</div></blockquote></div><br></div></div></div></blockquote></div><br></div=
></div></div>

--94eb2c082cbe7c72a0055d90cd29--