Mailing-List: contact user-help@flink.apache.org; run by ezmlm
Precedence: bulk
MIME-Version: 1.0
From: boci <boci.boci@gmail.com>
Date: Mon, 28 Aug 2017 21:29:41 +0000
Message-ID: <CAM8rHbGh-HWGjVVGRutoV+WPMiYXmwiz8dUX5rh1AExkj9NgbQ@mail.gmail.com>
Subject: Union limit
To: user@flink.apache.org
Content-Type: multipart/alternative; boundary="001a113fb358bbd9fe0557d6fec3"
archived-at: Mon, 28 Aug 2017 21:29:57 -0000

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

Hi guys!

I have one input (from mongo) and I split the incoming data to multiple
datasets (each created dynamically from configuration) and before I write
back the result I want to merge it to one dataset (there is some common
transformation).
so the flow:

DataSet from Mongod =>
Create Mappers dynamically (currently 74) so I have 74 DataSet =>
Custom filter and mapping on each dataset =>
Union dynamically to one (every mapper result is same type) =>
Some another common transformation =>
Count the result

but when I want to union more than 64 dataset I got these exception:

Exception in thread "main" org.apache.flink.optimizer.CompilerException:
Cannot currently handle nodes with more than 64 outputs.
at
org.apache.flink.optimizer.dag.OptimizerNode.addOutgoingConnection(OptimizerNode.java:348)
at
org.apache.flink.optimizer.dag.SingleInputNode.setInput(SingleInputNode.java:202)
at
org.apache.flink.optimizer.traversals.GraphCreatingVisitor.postVisit(GraphCreatingVisitor.java:268)
at
org.apache.flink.optimizer.traversals.GraphCreatingVisitor.postVisit(GraphCreatingVisitor.java:82)

I try to split the incoming (74) list of dataset to split to 60 + 14
 dataset and create an id mapper and union the result datasets but no
success:

val listOfDataSet: List[DataSet[...]] = ....

listOfDataSet
.sliding(60,60)
.map(dsg => dsg.reduce((ds1,ds2) => ds1.union(ds2)),map(new IdMapper()))
//There is an iterator of DataSet
.reduce((dsg1,dsg2) => dsg1.union(dsg2)) // Here I got the exception
.map(finalDataSet => ... some transformation ...)
.count()

There is any solution to solve this?

Thanks
b0c1

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

<div dir=3D"ltr">Hi guys!<div><br></div><div>I have one input (from mongo) =
and I split the incoming data to multiple datasets (each created dynamicall=
y from configuration) and before I write back the result I want to merge it=
 to one dataset (there is some common transformation).</div><div>so the flo=
w:</div><div><br></div><div>DataSet from Mongod =3D&gt;=C2=A0</div><div>Cre=
ate Mappers dynamically=C2=A0(currently 74) so I have 74 DataSet =3D&gt;=C2=
=A0</div><div>Custom filter and mapping on each dataset =3D&gt;=C2=A0</div>=
<div>Union dynamically to one (every mapper result is same type) =3D&gt;=C2=
=A0</div><div>Some another common transformation =3D&gt;</div><div>Count th=
e result</div><div><br></div><div>but when I want to union more than 64 dat=
aset I got these exception:</div><div><br></div><div><div>Exception in thre=
ad &quot;main&quot; org.apache.flink.optimizer.CompilerException: Cannot cu=
rrently handle nodes with more than 64 outputs.</div><div><span style=3D"wh=
ite-space:pre">	</span>at org.apache.flink.optimizer.dag.OptimizerNode.addO=
utgoingConnection(OptimizerNode.java:348)</div><div><span style=3D"white-sp=
ace:pre">	</span>at org.apache.flink.optimizer.dag.SingleInputNode.setInput=
(SingleInputNode.java:202)</div><div><span style=3D"white-space:pre">	</spa=
n>at org.apache.flink.optimizer.traversals.GraphCreatingVisitor.postVisit(G=
raphCreatingVisitor.java:268)</div><div><span style=3D"white-space:pre">	</=
span>at org.apache.flink.optimizer.traversals.GraphCreatingVisitor.postVisi=
t(GraphCreatingVisitor.java:82)</div></div><div><br></div><div>I try to spl=
it the incoming (74) list of dataset to split to 60=C2=A0+ 14 =C2=A0dataset=
 and create an id mapper and union the result datasets but no success:</div=
><div><br></div><div>val listOfDataSet: List[DataSet[...]] =3D ....</div><d=
iv><br>listOfDataSet<br>.sliding(60,60)<br>.map(dsg=C2=A0=3D&gt; dsg.reduce=
((ds1,ds2) =3D&gt; ds1.union(ds2)),map(new IdMapper()))</div><div>//There i=
s an iterator of DataSet</div><div>.reduce((dsg1,dsg2) =3D&gt; dsg1.union(d=
sg2)) // Here I got the exception<br>.map(finalDataSet =3D&gt; ... some tra=
nsformation ...)<br>.count()</div><div><br></div><div>There is any solution=
 to solve this?<br></div><div><br></div><div>Thanks</div><div>b0c1</div></d=
iv>

--001a113fb358bbd9fe0557d6fec3--