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: <CAAdrtT192vdunDQ5BEdUuORQtUJG_1evgvhii6SJTxkU9-=Njw@mail.gmail.com>
References: <CACDjLVgx9MTzXzncpEvrycQWdFh+AqTBRMyUd_8ACOyW6x4cvA@mail.gmail.com>
	<CAAdrtT0zEws+pxHe4S5RLAJMEY6OcwHWE5Z=igr8V_SASzFJjA@mail.gmail.com>
	<CACDjLViw+R2XshpmRP5HEH0Th8s-grY75K65UKjiWSKYX-MwXQ@mail.gmail.com>
	<CAAdrtT192vdunDQ5BEdUuORQtUJG_1evgvhii6SJTxkU9-=Njw@mail.gmail.com>
Date: Tue, 3 May 2016 15:13:03 +0200
Message-ID: <CACDjLVi09jsRwOG0V8vysjoLjv7fGAPXD-A-L+RvQ4S8R+e+DQ@mail.gmail.com>
Subject: Re: Creating a custom operator
From: Simone Robutti <simone.robutti@radicalbit.io>
To: user@flink.apache.org
Content-Type: multipart/alternative; boundary=001a1141176a7f2a7e0531efdef0
archived-at: Tue, 03 May 2016 13:13:10 -0000

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

I'm not sure this is the right way to do it but we were exploring all the
possibilities and this one is the more obvious. We also spent some time to
study how to do it to achieve a better understanding of Flink's internals.

What we want to do though is to integrate Flink with another distributed
system that builds its own nodes and coordinates through the network with
its own logic. This software is H2O (a Machine Learning platform) and the
integration consists of two big tasks: the first is to instantiate a H2O's
node in every task manager and handle the lifecycle of the node according
to the taskmanager and the execution graph. The second is to allow the
developer to code everything inside Flink, converting from and to H2O's
data structures (distributed tabular data) and triggering the execution of
algorithms on H2O with a uniform API.

Here's a simple example (assuming that we will use the TableAPI):

val env = ExecutionEnvironment.getExecutionEnvironment
val h2oEnv = H2OEnviroment.getEnvironment(env)

val myData: Table = ...
val someOtherData: Table = ...

val myH2OFrame = myData.select(...).toH2OFrame(h2oEnv)

val linearRegressionModel = h2oEnv.linearRegression(myH2OFrame)

val predictions:Table=linearRegressionModel(someOtherData)

predictions.select(...)


A good solution should allow the system to keep the H2O's nodes alive
through multiple tasks and the possibility to move the data locally from
Flink to H2O. The latter is not achieved in H2O's integration with Spark
but we still hope to do it.

That said, I'm still not sure if it is really required to implement a
custom runtime operator but given the complexity of the integration of two
distribute systems, we assumed that more control would allow more
flexibility and possibilities to achieve an ideal solution.


2016-05-03 13:29 GMT+02:00 Fabian Hueske <fhueske@gmail.com>:

> Hi Simone,
>
> you are right, the interfaces you extend are not considered to be public,
> user-facing API.
> Adding custom operators to the DataSet API touches many parts of the
> system and is not straightforward.
> The DataStream API has better support for custom operators.
>
> Can you explain what kind of operator you would like to add?
> Maybe the functionality can be achieved with the existing operators.
>
> Best, Fabian
>
> 2016-05-03 12:54 GMT+02:00 Simone Robutti <simone.robutti@radicalbit.io>:
>
>> Hello Fabian,
>>
>> we delved more moving from the input you gave us but a question arised.
>> We always assumed that runtime operators were open for extension without
>> modifying anything inside Flink but it looks like this is not the case and
>> the documentation assumes that the developer is working to a contribution
>> to Flink. So I would like to know if our understandment is correct and
>> custom runtime operators are not supposed to be implemented outside of
>> Flink.
>>
>> Thanks,
>>
>> Simone
>>
>> 2016-04-29 21:32 GMT+02:00 Fabian Hueske <fhueske@gmail.com>:
>>
>>> Hi Simone,
>>>
>>> the GraphCreatingVisitor transforms the common operator plan into a
>>> representation that is translated by the optimizer.
>>> You have to implement an OptimizerNode and OperatorDescriptor to
>>> describe the operator.
>>> Depending on the semantics of the operator, there are a few more places
>>> to make the integration working like driver strategies, cost model, etc.
>>>
>>> I would recommend to have a look at previous changes that added an
>>> operator such as PartitionOperator, SortPartitionOperator, OuterJoin, etc.
>>> The respective commits should give you an idea which parts of the code
>>> need to be touched. You should find the commit IDs in the JIRA issues for
>>> these extensions.
>>>
>>> Cheers, Fabian
>>>
>>>
>>>
>>>
>>>
>>> 2016-04-29 15:32 GMT+02:00 Simone Robutti <simone.robutti@radicalbit.io>
>>> :
>>>
>>>> Hello,
>>>>
>>>> I'm trying to create a custom operator to explore the internals of
>>>> Flink. Actually the one I'm working on is rather similar to Union and I'm
>>>> trying to mimick it for now. When I run my job though, this error arise:
>>>>
>>>> Exception in thread "main" java.lang.IllegalArgumentException: Unknown
>>>> operator type: MyOperator - My Operator
>>>> at
>>>> org.apache.flink.optimizer.traversals.GraphCreatingVisitor.preVisit(GraphCreatingVisitor.java:237)
>>>> at
>>>> org.apache.flink.optimizer.traversals.GraphCreatingVisitor.preVisit(GraphCreatingVisitor.java:82)
>>>> at
>>>> org.apache.flink.api.common.operators.DualInputOperator.accept(DualInputOperator.java:279)
>>>> at
>>>> org.apache.flink.api.common.operators.GenericDataSinkBase.accept(GenericDataSinkBase.java:223)
>>>> at org.apache.flink.api.common.Plan.accept(Plan.java:348)
>>>> at org.apache.flink.optimizer.Optimizer.compile(Optimizer.java:454)
>>>> at org.apache.flink.optimizer.Optimizer.compile(Optimizer.java:398)
>>>> at
>>>> org.apache.flink.client.LocalExecutor.getOptimizerPlanAsJSON(LocalExecutor.java:213)
>>>> at
>>>> org.apache.flink.api.java.LocalEnvironment.getExecutionPlan(LocalEnvironment.java:107)
>>>> at io.radicalbit.flinkh2o.Job$.main(Job.scala:50)
>>>> at io.radicalbit.flinkh2o.Job.main(Job.scala)
>>>> at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
>>>> at
>>>> sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:57)
>>>> at
>>>> sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
>>>> at java.lang.reflect.Method.invoke(Method.java:606)
>>>> at com.intellij.rt.execution.application.AppMain.main(AppMain.java:140)
>>>>
>>>> I looked at the location of the error but it's not clear to me how to
>>>> make my operator recognizable from the optimizer.
>>>>
>>>> Thank,
>>>>
>>>> Simone
>>>>
>>>
>>>
>>
>

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

<div dir=3D"ltr">I&#39;m not sure this is the right way to do it but we wer=
e exploring all the possibilities and this one is the more obvious. We also=
 spent some time to study how to do it to achieve a better understanding of=
 Flink&#39;s internals.<div><br></div><div>What we want to do though is to =
integrate Flink with another distributed system that builds its own nodes a=
nd coordinates through the network with its own logic. This software is H2O=
 (a Machine Learning platform) and the integration consists of two big task=
s: the first is to instantiate a H2O&#39;s node in every task manager and h=
andle the lifecycle of the node according to the taskmanager and the execut=
ion graph. The second is to allow the developer to code everything inside F=
link, converting from and to H2O&#39;s data structures (distributed tabular=
 data) and triggering the execution of algorithms on H2O with a uniform API=
.</div><div><br></div><div>Here&#39;s a simple example (assuming that we wi=
ll use the TableAPI):</div><div><br></div><div>val env =3D=C2=A0ExecutionEn=
vironment.getExecutionEnvironment</div><div>val h2oEnv =3D H2OEnviroment.ge=
tEnvironment(env)</div><div><br></div><div>val myData: Table =3D ...</div><=
div>val someOtherData: Table =3D ...</div><div><br></div><div>val myH2OFram=
e =3D myData.select(...).toH2OFrame(h2oEnv)</div><div><br></div><div>val li=
nearRegressionModel =3D h2oEnv.linearRegression(myH2OFrame)</div><div><br><=
/div><div>val predictions:Table=3DlinearRegressionModel(someOtherData)</div=
><div><br></div><div>predictions.select(...)</div><div><br></div><div><br><=
/div><div>A good solution should allow the system to keep the H2O&#39;s nod=
es alive through multiple tasks and the possibility to move the data locall=
y from Flink to H2O. The latter is not achieved in H2O&#39;s integration wi=
th Spark but we still hope to do it.</div><div><br></div><div>That said, I&=
#39;m still not sure if it is really required to implement a custom runtime=
 operator but given the complexity of the integration of two distribute sys=
tems, we assumed that more control would allow more flexibility and possibi=
lities to achieve an ideal solution.</div><div><br></div><div><br></div><di=
v><br></div><div><br></div></div><div class=3D"gmail_extra"><br><div class=
=3D"gmail_quote">2016-05-03 13:29 GMT+02:00 Fabian Hueske <span dir=3D"ltr"=
>&lt;<a href=3D"mailto:fhueske@gmail.com" target=3D"_blank">fhueske@gmail.c=
om</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 dir=3D"ltr"><div>=
<div><div><div>Hi Simone,<br><br></div>you are right, the interfaces you ex=
tend are not considered to be public, user-facing API. <br>Adding custom op=
erators to the DataSet API touches many parts of the system and is not stra=
ightforward.<br></div>The DataStream API has better support for custom oper=
ators.<br><br></div>Can you explain what kind of operator you would like to=
 add? <br>Maybe the functionality can be achieved with the existing operato=
rs.<br><br></div>Best, Fabian<br></div><div class=3D"HOEnZb"><div class=3D"=
h5"><div class=3D"gmail_extra"><br><div class=3D"gmail_quote">2016-05-03 12=
:54 GMT+02:00 Simone Robutti <span dir=3D"ltr">&lt;<a href=3D"mailto:simone=
.robutti@radicalbit.io" target=3D"_blank">simone.robutti@radicalbit.io</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 dir=3D"ltr">Hello Fabian=
,<div><br></div><div>we delved more moving from the input you gave us but a=
 question arised. We always assumed that runtime operators were open for ex=
tension without modifying anything inside Flink but it looks like this is n=
ot the case and the documentation assumes that the developer is working to =
a contribution to Flink. So I would like to know if our understandment is c=
orrect and custom runtime operators are not supposed to be implemented outs=
ide of Flink.</div><div><br></div><div>Thanks,</div><div><br></div><div>Sim=
one</div></div><div><div><div class=3D"gmail_extra"><br><div class=3D"gmail=
_quote">2016-04-29 21:32 GMT+02:00 Fabian Hueske <span dir=3D"ltr">&lt;<a h=
ref=3D"mailto:fhueske@gmail.com" target=3D"_blank">fhueske@gmail.com</a>&gt=
;</span>:<br><blockquote class=3D"gmail_quote" style=3D"margin:0 0 0 .8ex;b=
order-left:1px #ccc solid;padding-left:1ex"><div dir=3D"ltr"><div><div><div=
><div>Hi Simone,<br><br></div><div>the GraphCreatingVisitor transforms the =
common operator plan into a representation that is translated by the optimi=
zer.<br>You have to implement an OptimizerNode and OperatorDescriptor to de=
scribe the operator.<br></div><div>Depending on the semantics of the operat=
or, there are a few more places to make the integration working like driver=
 strategies, cost model, etc.<br></div><div><br>I would recommend to have a=
 look at previous changes that added an operator such as PartitionOperator,=
 SortPartitionOperator, OuterJoin, etc.<br></div><div>The respective commit=
s should give you an idea which parts of the code need to be touched. You s=
hould find the commit IDs in the JIRA issues for these extensions.<br><br><=
/div><div>Cheers, Fabian<br></div><div><br></div><div><br></div></div></div=
></div><div><div><div><div><br><br></div></div></div></div></div><div><div>=
<div class=3D"gmail_extra"><br><div class=3D"gmail_quote">2016-04-29 15:32 =
GMT+02:00 Simone Robutti <span dir=3D"ltr">&lt;<a href=3D"mailto:simone.rob=
utti@radicalbit.io" target=3D"_blank">simone.robutti@radicalbit.io</a>&gt;<=
/span>:<br><blockquote class=3D"gmail_quote" style=3D"margin:0 0 0 .8ex;bor=
der-left:1px #ccc solid;padding-left:1ex"><div dir=3D"ltr">Hello,<div><br><=
/div><div>I&#39;m trying to create a custom operator to explore the interna=
ls of Flink. Actually the one I&#39;m working on is rather similar to Union=
 and I&#39;m trying to mimick it for now. When I run my job though, this er=
ror arise:</div><div><br></div><div><div>Exception in thread &quot;main&quo=
t; java.lang.IllegalArgumentException: Unknown operator type: MyOperator - =
My Operator</div><div><span style=3D"white-space:pre-wrap">	</span>at org.a=
pache.flink.optimizer.traversals.GraphCreatingVisitor.preVisit(GraphCreatin=
gVisitor.java:237)</div><div><span style=3D"white-space:pre-wrap">	</span>a=
t org.apache.flink.optimizer.traversals.GraphCreatingVisitor.preVisit(Graph=
CreatingVisitor.java:82)</div><div><span style=3D"white-space:pre-wrap">	</=
span>at org.apache.flink.api.common.operators.DualInputOperator.accept(Dual=
InputOperator.java:279)</div><div><span style=3D"white-space:pre-wrap">	</s=
pan>at org.apache.flink.api.common.operators.GenericDataSinkBase.accept(Gen=
ericDataSinkBase.java:223)</div><div><span style=3D"white-space:pre-wrap">	=
</span>at org.apache.flink.api.common.Plan.accept(Plan.java:348)</div><div>=
<span style=3D"white-space:pre-wrap">	</span>at org.apache.flink.optimizer.=
Optimizer.compile(Optimizer.java:454)</div><div><span style=3D"white-space:=
pre-wrap">	</span>at org.apache.flink.optimizer.Optimizer.compile(Optimizer=
.java:398)</div><div><span style=3D"white-space:pre-wrap">	</span>at org.ap=
ache.flink.client.LocalExecutor.getOptimizerPlanAsJSON(LocalExecutor.java:2=
13)</div><div><span style=3D"white-space:pre-wrap">	</span>at org.apache.fl=
ink.api.java.LocalEnvironment.getExecutionPlan(LocalEnvironment.java:107)</=
div><div><span style=3D"white-space:pre-wrap">	</span>at io.radicalbit.flin=
kh2o.Job$.main(Job.scala:50)</div><div><span style=3D"white-space:pre-wrap"=
>	</span>at io.radicalbit.flinkh2o.Job.main(Job.scala)</div><div><span styl=
e=3D"white-space:pre-wrap">	</span>at sun.reflect.NativeMethodAccessorImpl.=
invoke0(Native Method)</div><div><span style=3D"white-space:pre-wrap">	</sp=
an>at sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.=
java:57)</div><div><span style=3D"white-space:pre-wrap">	</span>at sun.refl=
ect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:4=
3)</div><div><span style=3D"white-space:pre-wrap">	</span>at java.lang.refl=
ect.Method.invoke(Method.java:606)</div><div><span style=3D"white-space:pre=
-wrap">	</span>at com.intellij.rt.execution.application.AppMain.main(AppMai=
n.java:140)</div></div><div><br></div><div>I looked at the location of the =
error but it&#39;s not clear to me how to make my operator recognizable fro=
m the optimizer.=C2=A0</div><div><br></div><div>Thank,</div><div><br></div>=
<div>Simone</div></div>
</blockquote></div><br></div>
</div></div></blockquote></div><br></div>
</div></div></blockquote></div><br></div>
</div></div></blockquote></div><br></div>

--001a1141176a7f2a7e0531efdef0--