Mailing-List: contact user-help@flink.apache.org; run by ezmlm
Precedence: bulk
Reply-To: user@flink.apache.org
MIME-Version: 1.0
In-Reply-To: <CAAqdGsfD6o_W77hkkr4p98MEcD=+k3Z5qe7roQtnmVCJQ0Gxrg@mail.gmail.com>
References: <CAAqdGsfD6o_W77hkkr4p98MEcD=+k3Z5qe7roQtnmVCJQ0Gxrg@mail.gmail.com>
From: Fabian Hueske <fhueske@gmail.com>
Date: Mon, 28 Nov 2016 10:02:28 +0100
Message-ID: <CAAdrtT1OZW6e5uWjfovkni6whXkK_XApjHvuexwbiSqXYq=dqg@mail.gmail.com>
Subject: Re: DB connection and query inside map function
To: user@flink.apache.org
Content-Type: multipart/alternative; boundary=001a11408008fce6fa054258bc5a
archived-at: Mon, 28 Nov 2016 09:03:10 -0000

--001a11408008fce6fa054258bc5a
Content-Type: text/plain; charset=UTF-8

Hi Anastasios,

that's certainly possible. The most straight-forward approach would be a
synchronous call to the database.
Because only one request is active at the same time, you do not need a
thread pool.
You can establish the connection in the open() method of a RichMapFunction.
The problem with this approach is that the synchronous requests can
significantly increase the latency.

Doing the calls asynchronously and using a thread pool is not very easy
because this would need to be integrated with Flink's checkpointing
mechanism.
In fact, there is an effort to add a special Map operator that supports
asynchronous calls (see FLIP-12 [1]).
We expect this to be included in the next minor release, Flink 1.2.

Hope this helps,
Fabian

[1]
https://cwiki.apache.org/confluence/pages/viewpage.action?pageId=65870673

2016-11-27 22:11 GMT+01:00 Anastasios Skarlatidis <a.skarlatidis@gmail.com>:

> Hi!
>
> I am new to Apache Flink and I would like to ask what is the best way to
> query a relational DB inside a map function, in order to enrich the
> streaming data. Consider, for example, that I have a KeyedStream[Int,
> String] and I would like to query the database based on the Int value
> inside a map function `stream.map(v: Int => <<some SQL query>> )`.
>
> Is it possible to have a connection pooler per worker nod,e in order to be
> used inside each map function call?
>
> Best,
>
> Anastasios
>

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

<div dir=3D"ltr"><div><div><div><div>Hi Anastasios,<br><br></div>that&#39;s=
 certainly possible. The most straight-forward approach would be a synchron=
ous call to the database.<br></div><div>Because only one request is active =
at the same time, you do not need a thread pool.<br></div>You can establish=
 the connection in the open() method of a RichMapFunction. The problem with=
 this approach is that the synchronous requests can significantly increase =
the latency.<br><br></div><div>Doing the calls asynchronously and using a t=
hread pool is not very easy because this would need to be integrated with F=
link&#39;s checkpointing mechanism.<br></div><div>In fact, there is an effo=
rt to add a special Map operator that supports asynchronous calls (see FLIP=
-12 [1]).<br></div><div>We expect this to be included in the next minor rel=
ease, Flink 1.2.<br><br></div><div>Hope this helps,<br></div><div>Fabian<br=
></div></div><div><div><div><br>[1] <a href=3D"https://cwiki.apache.org/con=
fluence/pages/viewpage.action?pageId=3D65870673">https://cwiki.apache.org/c=
onfluence/pages/viewpage.action?pageId=3D65870673</a><br></div></div></div>=
</div><div class=3D"gmail_extra"><br><div class=3D"gmail_quote">2016-11-27 =
22:11 GMT+01:00 Anastasios Skarlatidis <span dir=3D"ltr">&lt;<a href=3D"mai=
lto:a.skarlatidis@gmail.com" target=3D"_blank">a.skarlatidis@gmail.com</a>&=
gt;</span>:<br><blockquote class=3D"gmail_quote" style=3D"margin:0 0 0 .8ex=
;border-left:1px #ccc solid;padding-left:1ex"><div style=3D"word-wrap:break=
-word"><div id=3D"m_7353981749416225783bloop_customfont" style=3D"font-fami=
ly:Helvetica,Arial;font-size:13px;color:rgba(0,0,0,1.0);margin:0px;line-hei=
ght:auto">Hi!=C2=A0</div><div id=3D"m_7353981749416225783bloop_sign_1480280=
939919121920" class=3D"m_7353981749416225783bloop_sign"><div id=3D"m_735398=
1749416225783bloop_sign_1480280939919121920" class=3D"m_7353981749416225783=
bloop_sign"><br></div><div id=3D"m_7353981749416225783bloop_sign_1480280939=
919121920" class=3D"m_7353981749416225783bloop_sign">I am new to Apache Fli=
nk and I would like to ask what is the best way to query a relational DB in=
side a map function, in order to enrich the streaming data. Consider, for e=
xample, that I have a KeyedStream[Int, String] and I would like to query th=
e database based on the Int value inside a map function `stream.map(v: Int =
=3D&gt; &lt;&lt;some SQL query&gt;&gt; )`.=C2=A0</div><div id=3D"m_73539817=
49416225783bloop_sign_1480280939919121920" class=3D"m_7353981749416225783bl=
oop_sign"><br></div><div id=3D"m_7353981749416225783bloop_sign_148028093991=
9121920" class=3D"m_7353981749416225783bloop_sign">Is it possible to have a=
 connection pooler per worker nod,e in order to be used inside each map fun=
ction call?</div><div id=3D"m_7353981749416225783bloop_sign_148028093991912=
1920" class=3D"m_7353981749416225783bloop_sign"><br></div><div id=3D"m_7353=
981749416225783bloop_sign_1480280939919121920" class=3D"m_73539817494162257=
83bloop_sign">Best,</div><div id=3D"m_7353981749416225783bloop_sign_1480280=
939919121920" class=3D"m_7353981749416225783bloop_sign"><br></div><div id=
=3D"m_7353981749416225783bloop_sign_1480280939919121920" class=3D"m_7353981=
749416225783bloop_sign">Anastasios</div></div></div>
</blockquote></div><br></div>

--001a11408008fce6fa054258bc5a--