Mailing-List: contact user-help@flink.apache.org; run by ezmlm
Precedence: bulk
Reply-To: user@flink.apache.org
MIME-Version: 1.0
From: Yury Ruchin <yuri.ruchin@gmail.com>
Date: Tue, 20 Dec 2016 12:50:10 +0300
Message-ID: <CACVLuZw6iRkX20m12CCAAKsQoVoWU1YC50JPv+4m4meHFbgTRw@mail.gmail.com>
Subject: Implementing tee functionality in a streaming job
To: user@flink.apache.org
Content-Type: multipart/alternative; boundary=94eb2c074c0e54d1ab054413f631
archived-at: Tue, 20 Dec 2016 09:50:21 -0000

--94eb2c074c0e54d1ab054413f631
Content-Type: text/plain; charset=UTF-8

Hi all,

I have a streaming job that I want at some point to duplicate stream, so
that one copy of it ends up in a sink, and another one goes down the
processing pipeline. This way I want to implement something similar to
"tee" Linux utility.

I tried a naive approach that looked like this:

val streamToMirror = env.addSource(mySource).<some operators here>
streamToMirror.addSink(someSink) // (1) tee
streamToMirror.<more operators here>.addSink(anotherSink) // (2) process
further

But the above results in the stream being split between the mySink (1) and
further processors (2) in a seemingly nondeterministic way.

I could write through Kafka as shown in this pseudocode:

val headStream = env.addSource(mySource).<some operators here>
headStream.addSink(KafkaSink("myTopic"))
val tailStream = env.addSource(KafkaSource("myTopic")).<more operators here>

But this would incur additional latency + deserialization overhead that I
would like to avoid.

Is there any way to implement stream teeing as described?

Thanks,
Yury

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

<div dir=3D"ltr">Hi all,<div><br></div><div>I have a streaming job that I w=
ant at some point to duplicate stream, so that one copy of it ends up in a =
sink, and another one goes down the processing pipeline. This way I want to=
 implement something similar to &quot;tee&quot; Linux utility.</div><div><b=
r></div><div>I tried a naive approach that looked like this:</div><div><br>=
</div><div><font face=3D"monospace, monospace">val streamToMirror =3D env.a=
ddSource(mySource).&lt;some operators here&gt;</font></div><div><font face=
=3D"monospace, monospace">streamToMirror.addSink(someSink) // (1) tee=C2=A0=
</font></div><div><font face=3D"monospace, monospace">streamToMirror.&lt;mo=
re operators here&gt;.addSink(anotherSink) // (2) process further</font></d=
iv><div><br></div><div>But the above results in the stream being split betw=
een the mySink (1) and further processors (2) in a seemingly nondeterminist=
ic way.</div><div><br></div><div>I could write through Kafka as shown in th=
is pseudocode:</div><div><br></div><div><font face=3D"monospace, monospace"=
>val headStream =3D env.addSource(mySource).&lt;some operators here&gt;</fo=
nt></div><div><font face=3D"monospace, monospace">headStream.addSink(KafkaS=
ink(&quot;myTopic&quot;))</font></div><div><font face=3D"monospace, monospa=
ce">val tailStream =3D env.addSource(KafkaSource(&quot;myTopic&quot;)).&lt;=
more operators here&gt;</font></div><div><br></div><div>But this would incu=
r additional latency + deserialization overhead that I would like to avoid.=
</div><div><br></div><div>Is there any way to implement stream teeing as de=
scribed?</div><div><br></div><div>Thanks,</div><div>Yury</div></div>

--94eb2c074c0e54d1ab054413f631--