Re: Flink Data Stream Union
Do you really need to iterate ? On Mon, Oct 19, 2015 at 5:42 PM, flinkuser wrote: > > Here is my code snippet but I find the union operator not workable. > > DataStream msgDataStream1 = env.addSource((new > SocketSource(hostName1,port,'\n',-1))).filter(new > MessageFilter()).setParallelism(1); > DataStream msgDataStream2 = env.addSource((new > SocketSource(hostName2,port,'\n',-1))).filter(new > MessageFilter()).setParallelism(1); > > > DataStream stockStream = > (msgDataStream1.union(msgDataStream2)).iterate(); > stockStream.print(); > > > The stockStream doesn’t print the consolidated stream data. Sometimes > Stream1 is printed, sometimes none is printed. > > Can you please help me out, as of what is wrong here. > > > > -- > View this message in context: > http://apache-flink-user-mailing-list-archive.2336050.n4.nabble.com/Flink-Data-Stream-Union-tp3169.html > Sent from the Apache Flink User Mailing List archive. mailing list archive > at Nabble.com. >
Re: Flink Data Stream Union
Hi! Two comments: (1) The iterate() statement is probably wrong, as noticed by Anwar. (2) Which version of Flink are you using? In 0.9.x, the Union operator is not lock-safe, in 0.10, it should work well. The 0.10 release is coming up shortly, you can try the 0.10-SNAPSHOT version already. Greetings, Stephan On Mon, Oct 19, 2015 at 6:01 PM, Anwar Rizal wrote: > Do you really need to iterate ? > > On Mon, Oct 19, 2015 at 5:42 PM, flinkuser wrote: > >> >> Here is my code snippet but I find the union operator not workable. >> >> DataStream msgDataStream1 = env.addSource((new >> SocketSource(hostName1,port,'\n',-1))).filter(new >> MessageFilter()).setParallelism(1); >> DataStream msgDataStream2 = env.addSource((new >> SocketSource(hostName2,port,'\n',-1))).filter(new >> MessageFilter()).setParallelism(1); >> >> >> DataStream stockStream = >> (msgDataStream1.union(msgDataStream2)).iterate(); >> stockStream.print(); >> >> >> The stockStream doesn’t print the consolidated stream data. Sometimes >> Stream1 is printed, sometimes none is printed. >> >> Can you please help me out, as of what is wrong here. >> >> >> >> -- >> View this message in context: >> http://apache-flink-user-mailing-list-archive.2336050.n4.nabble.com/Flink-Data-Stream-Union-tp3169.html >> Sent from the Apache Flink User Mailing List archive. mailing list >> archive at Nabble.com. >> > >
Re: Flink Data Stream Union
Here is the strange behavior.
Below code works in one box but not in the other. I had it working in my
laptop the whole of yesterday, but strangely today it doesnt work in my
desktop.
Can anyone please let me know what the issue is.
public static void main(String[] args) throws Exception {
try {
final StreamExecutionEnvironment env =
StreamExecutionEnvironment.getExecutionEnvironment();
DataStream msgDataStream1 = env.addSource((new
SocketSource(hostName1, port, '\n', -1))).filter(new
MessageFilter()).setParallelism(1);
DataStream msgDataStream2 = env.addSource((new
SocketSource(hostName2, port, '\n', -1))).filter(new
MessageFilter()).setParallelism(1);
env.execute("Stock stream");
} catch (Exception e) {
System.err.println("Exception = > " + e.getMessage());
e.printStackTrace();
}
}
private static void unionMessageStreams(DataStream
msgDataStream1,
DataStream msgDataStream2) {
try {
DataStream ds =
msgDataStream1.union(msgDataStream2);
ds.print();
} catch (Exception e) {
System.err.println("Exception in union Message Streams
() = > " +
e.getMessage());
}
}
Thanks
--
View this message in context:
http://apache-flink-user-mailing-list-archive.2336050.n4.nabble.com/Flink-Data-Stream-Union-tp3169p3196.html
Sent from the Apache Flink User Mailing List archive. mailing list archive at
Nabble.com.
Re: Flink Data Stream Union
Can it be that you forgot to call unionMessageStreams in your main method?
Cheers,
Till
On Wed, Oct 21, 2015 at 3:02 PM, flinkuser wrote:
> Here is the strange behavior.
>
> Below code works in one box but not in the other. I had it working in my
> laptop the whole of yesterday, but strangely today it doesnt work in my
> desktop.
>
> Can anyone please let me know what the issue is.
>
>
> public static void main(String[] args) throws Exception {
> try {
> final StreamExecutionEnvironment env =
> StreamExecutionEnvironment.getExecutionEnvironment();
> DataStream msgDataStream1 =
> env.addSource((new
> SocketSource(hostName1, port, '\n', -1))).filter(new
> MessageFilter()).setParallelism(1);
> DataStream msgDataStream2 =
> env.addSource((new
> SocketSource(hostName2, port, '\n', -1))).filter(new
> MessageFilter()).setParallelism(1);
>
> env.execute("Stock stream");
>
> } catch (Exception e) {
> System.err.println("Exception = > " +
> e.getMessage());
> e.printStackTrace();
> }
> }
>
> private static void unionMessageStreams(DataStream
> msgDataStream1,
> DataStream msgDataStream2) {
> try {
>
> DataStream ds =
> msgDataStream1.union(msgDataStream2);
> ds.print();
> } catch (Exception e) {
> System.err.println("Exception in union Message
> Streams () = > " +
> e.getMessage());
> }
> }
>
> Thanks
>
>
>
> --
> View this message in context:
> http://apache-flink-user-mailing-list-archive.2336050.n4.nabble.com/Flink-Data-Stream-Union-tp3169p3196.html
> Sent from the Apache Flink User Mailing List archive. mailing list archive
> at Nabble.com.
>
Re: Flink Data Stream Union
Hi Gayu,
could it be that no data ever arrives on the second input stream? Or that the
filter filters out all messages?
Also, in the example you posted you forgot to call unionMessageStreams().
Cheers,
Aljoscha
> On 21 Oct 2015, at 15:29, Till Rohrmann wrote:
>
> Can it be that you forgot to call unionMessageStreams in your main method?
>
> Cheers,
> Till
>
>
> On Wed, Oct 21, 2015 at 3:02 PM, flinkuser wrote:
> Here is the strange behavior.
>
> Below code works in one box but not in the other. I had it working in my
> laptop the whole of yesterday, but strangely today it doesnt work in my
> desktop.
>
> Can anyone please let me know what the issue is.
>
>
> public static void main(String[] args) throws Exception {
> try {
> final StreamExecutionEnvironment env =
> StreamExecutionEnvironment.getExecutionEnvironment();
> DataStream msgDataStream1 = env.addSource((new
> SocketSource(hostName1, port, '\n', -1))).filter(new
> MessageFilter()).setParallelism(1);
> DataStream msgDataStream2 = env.addSource((new
> SocketSource(hostName2, port, '\n', -1))).filter(new
> MessageFilter()).setParallelism(1);
>
> env.execute("Stock stream");
>
> } catch (Exception e) {
> System.err.println("Exception = > " +
> e.getMessage());
> e.printStackTrace();
> }
> }
>
> private static void unionMessageStreams(DataStream
> msgDataStream1,
> DataStream msgDataStream2) {
> try {
>
> DataStream ds =
> msgDataStream1.union(msgDataStream2);
> ds.print();
> } catch (Exception e) {
> System.err.println("Exception in union Message
> Streams () = > " +
> e.getMessage());
> }
> }
>
> Thanks
>
>
>
> --
> View this message in context:
> http://apache-flink-user-mailing-list-archive.2336050.n4.nabble.com/Flink-Data-Stream-Union-tp3169p3196.html
> Sent from the Apache Flink User Mailing List archive. mailing list archive at
> Nabble.com.
>
Re: Flink Data Stream Union
The data does arrive in the second port and i am able to see that in the
filter class received.
It happens only on specific machine on which i run the code.
Yes, i did forget to post here, but my program calls the
unionMessageStreams()
On Wed, Oct 21, 2015 at 9:39 AM, Aljoscha Krettek <
aljoscha.kret...@gmail.com> wrote:
> Hi Gayu,
> could it be that no data ever arrives on the second input stream? Or that
> the filter filters out all messages?
>
> Also, in the example you posted you forgot to call unionMessageStreams().
>
> Cheers,
> Aljoscha
>
> > On 21 Oct 2015, at 15:29, Till Rohrmann wrote:
> >
> > Can it be that you forgot to call unionMessageStreams in your main
> method?
> >
> > Cheers,
> > Till
> >
> >
> > On Wed, Oct 21, 2015 at 3:02 PM, flinkuser wrote:
> > Here is the strange behavior.
> >
> > Below code works in one box but not in the other. I had it working in my
> > laptop the whole of yesterday, but strangely today it doesnt work in my
> > desktop.
> >
> > Can anyone please let me know what the issue is.
> >
> >
> > public static void main(String[] args) throws Exception {
> > try {
> > final StreamExecutionEnvironment env =
> > StreamExecutionEnvironment.getExecutionEnvironment();
> > DataStream msgDataStream1 =
> env.addSource((new
> > SocketSource(hostName1, port, '\n', -1))).filter(new
> > MessageFilter()).setParallelism(1);
> > DataStream msgDataStream2 =
> env.addSource((new
> > SocketSource(hostName2, port, '\n', -1))).filter(new
> > MessageFilter()).setParallelism(1);
> >
> > env.execute("Stock stream");
> >
> > } catch (Exception e) {
> > System.err.println("Exception = > " +
> e.getMessage());
> > e.printStackTrace();
> > }
> > }
> >
> > private static void unionMessageStreams(DataStream
> msgDataStream1,
> > DataStream msgDataStream2) {
> > try {
> >
> > DataStream ds =
> msgDataStream1.union(msgDataStream2);
> > ds.print();
> > } catch (Exception e) {
> > System.err.println("Exception in union Message
> Streams () = > " +
> > e.getMessage());
> > }
> > }
> >
> > Thanks
> >
> >
> >
> > --
> > View this message in context:
> http://apache-flink-user-mailing-list-archive.2336050.n4.nabble.com/Flink-Data-Stream-Union-tp3169p3196.html
> > Sent from the Apache Flink User Mailing List archive. mailing list
> archive at Nabble.com.
> >
>
>
--
Gayu
Re: Flink Data Stream Union
So it is received in the filter but the print afterwards does not print?
> On 21 Oct 2015, at 15:49, Gayu wrote:
>
> The data does arrive in the second port and i am able to see that in the
> filter class received.
> It happens only on specific machine on which i run the code.
>
>
> Yes, i did forget to post here, but my program calls the unionMessageStreams()
>
> On Wed, Oct 21, 2015 at 9:39 AM, Aljoscha Krettek
> wrote:
> Hi Gayu,
> could it be that no data ever arrives on the second input stream? Or that the
> filter filters out all messages?
>
> Also, in the example you posted you forgot to call unionMessageStreams().
>
> Cheers,
> Aljoscha
>
> > On 21 Oct 2015, at 15:29, Till Rohrmann wrote:
> >
> > Can it be that you forgot to call unionMessageStreams in your main method?
> >
> > Cheers,
> > Till
> >
> >
> > On Wed, Oct 21, 2015 at 3:02 PM, flinkuser wrote:
> > Here is the strange behavior.
> >
> > Below code works in one box but not in the other. I had it working in my
> > laptop the whole of yesterday, but strangely today it doesnt work in my
> > desktop.
> >
> > Can anyone please let me know what the issue is.
> >
> >
> > public static void main(String[] args) throws Exception {
> > try {
> > final StreamExecutionEnvironment env =
> > StreamExecutionEnvironment.getExecutionEnvironment();
> > DataStream msgDataStream1 =
> > env.addSource((new
> > SocketSource(hostName1, port, '\n', -1))).filter(new
> > MessageFilter()).setParallelism(1);
> > DataStream msgDataStream2 =
> > env.addSource((new
> > SocketSource(hostName2, port, '\n', -1))).filter(new
> > MessageFilter()).setParallelism(1);
> >
> > env.execute("Stock stream");
> >
> > } catch (Exception e) {
> > System.err.println("Exception = > " +
> > e.getMessage());
> > e.printStackTrace();
> > }
> > }
> >
> > private static void unionMessageStreams(DataStream
> > msgDataStream1,
> > DataStream msgDataStream2) {
> > try {
> >
> > DataStream ds =
> > msgDataStream1.union(msgDataStream2);
> > ds.print();
> > } catch (Exception e) {
> > System.err.println("Exception in union Message
> > Streams () = > " +
> > e.getMessage());
> > }
> > }
> >
> > Thanks
> >
> >
> >
> > --
> > View this message in context:
> > http://apache-flink-user-mailing-list-archive.2336050.n4.nabble.com/Flink-Data-Stream-Union-tp3169p3196.html
> > Sent from the Apache Flink User Mailing List archive. mailing list archive
> > at Nabble.com.
> >
>
>
>
>
> --
> Gayu
Re: Flink Data Stream Union
Yes, exactly.
On Wed, Oct 21, 2015 at 10:17 AM, Aljoscha Krettek <
aljoscha.kret...@gmail.com> wrote:
> So it is received in the filter but the print afterwards does not print?
> > On 21 Oct 2015, at 15:49, Gayu wrote:
> >
> > The data does arrive in the second port and i am able to see that in the
> filter class received.
> > It happens only on specific machine on which i run the code.
> >
> >
> > Yes, i did forget to post here, but my program calls the
> unionMessageStreams()
> >
> > On Wed, Oct 21, 2015 at 9:39 AM, Aljoscha Krettek <
> aljoscha.kret...@gmail.com> wrote:
> > Hi Gayu,
> > could it be that no data ever arrives on the second input stream? Or
> that the filter filters out all messages?
> >
> > Also, in the example you posted you forgot to call unionMessageStreams().
> >
> > Cheers,
> > Aljoscha
> >
> > > On 21 Oct 2015, at 15:29, Till Rohrmann wrote:
> > >
> > > Can it be that you forgot to call unionMessageStreams in your main
> method?
> > >
> > > Cheers,
> > > Till
> > >
> > >
> > > On Wed, Oct 21, 2015 at 3:02 PM, flinkuser wrote:
> > > Here is the strange behavior.
> > >
> > > Below code works in one box but not in the other. I had it working in
> my
> > > laptop the whole of yesterday, but strangely today it doesnt work in my
> > > desktop.
> > >
> > > Can anyone please let me know what the issue is.
> > >
> > >
> > > public static void main(String[] args) throws Exception {
> > > try {
> > > final StreamExecutionEnvironment env =
> > > StreamExecutionEnvironment.getExecutionEnvironment();
> > > DataStream msgDataStream1 =
> env.addSource((new
> > > SocketSource(hostName1, port, '\n', -1))).filter(new
> > > MessageFilter()).setParallelism(1);
> > > DataStream msgDataStream2 =
> env.addSource((new
> > > SocketSource(hostName2, port, '\n', -1))).filter(new
> > > MessageFilter()).setParallelism(1);
> > >
> > > env.execute("Stock stream");
> > >
> > > } catch (Exception e) {
> > > System.err.println("Exception = > " +
> e.getMessage());
> > > e.printStackTrace();
> > > }
> > > }
> > >
> > > private static void unionMessageStreams(DataStream
> msgDataStream1,
> > > DataStream msgDataStream2) {
> > > try {
> > >
> > > DataStream ds =
> msgDataStream1.union(msgDataStream2);
> > > ds.print();
> > > } catch (Exception e) {
> > > System.err.println("Exception in union Message
> Streams () = > " +
> > > e.getMessage());
> > > }
> > > }
> > >
> > > Thanks
> > >
> > >
> > >
> > > --
> > > View this message in context:
> http://apache-flink-user-mailing-list-archive.2336050.n4.nabble.com/Flink-Data-Stream-Union-tp3169p3196.html
> > > Sent from the Apache Flink User Mailing List archive. mailing list
> archive at Nabble.com.
> > >
> >
> >
> >
> >
> > --
> > Gayu
>
>
--
Gayu
Re: Flink Data Stream Union
So does the filter maybe filter out everything?
> On 21 Oct 2015, at 16:18, Gayu wrote:
>
> Yes, exactly.
>
> On Wed, Oct 21, 2015 at 10:17 AM, Aljoscha Krettek
> wrote:
> So it is received in the filter but the print afterwards does not print?
> > On 21 Oct 2015, at 15:49, Gayu wrote:
> >
> > The data does arrive in the second port and i am able to see that in the
> > filter class received.
> > It happens only on specific machine on which i run the code.
> >
> >
> > Yes, i did forget to post here, but my program calls the
> > unionMessageStreams()
> >
> > On Wed, Oct 21, 2015 at 9:39 AM, Aljoscha Krettek
> > wrote:
> > Hi Gayu,
> > could it be that no data ever arrives on the second input stream? Or that
> > the filter filters out all messages?
> >
> > Also, in the example you posted you forgot to call unionMessageStreams().
> >
> > Cheers,
> > Aljoscha
> >
> > > On 21 Oct 2015, at 15:29, Till Rohrmann wrote:
> > >
> > > Can it be that you forgot to call unionMessageStreams in your main method?
> > >
> > > Cheers,
> > > Till
> > >
> > >
> > > On Wed, Oct 21, 2015 at 3:02 PM, flinkuser wrote:
> > > Here is the strange behavior.
> > >
> > > Below code works in one box but not in the other. I had it working in my
> > > laptop the whole of yesterday, but strangely today it doesnt work in my
> > > desktop.
> > >
> > > Can anyone please let me know what the issue is.
> > >
> > >
> > > public static void main(String[] args) throws Exception {
> > > try {
> > > final StreamExecutionEnvironment env =
> > > StreamExecutionEnvironment.getExecutionEnvironment();
> > > DataStream msgDataStream1 =
> > > env.addSource((new
> > > SocketSource(hostName1, port, '\n', -1))).filter(new
> > > MessageFilter()).setParallelism(1);
> > > DataStream msgDataStream2 =
> > > env.addSource((new
> > > SocketSource(hostName2, port, '\n', -1))).filter(new
> > > MessageFilter()).setParallelism(1);
> > >
> > > env.execute("Stock stream");
> > >
> > > } catch (Exception e) {
> > > System.err.println("Exception = > " +
> > > e.getMessage());
> > > e.printStackTrace();
> > > }
> > > }
> > >
> > > private static void unionMessageStreams(DataStream
> > > msgDataStream1,
> > > DataStream msgDataStream2) {
> > > try {
> > >
> > > DataStream ds =
> > > msgDataStream1.union(msgDataStream2);
> > > ds.print();
> > > } catch (Exception e) {
> > > System.err.println("Exception in union Message
> > > Streams () = > " +
> > > e.getMessage());
> > > }
> > > }
> > >
> > > Thanks
> > >
> > >
> > >
> > > --
> > > View this message in context:
> > > http://apache-flink-user-mailing-list-archive.2336050.n4.nabble.com/Flink-Data-Stream-Union-tp3169p3196.html
> > > Sent from the Apache Flink User Mailing List archive. mailing list
> > > archive at Nabble.com.
> > >
> >
> >
> >
> >
> > --
> > Gayu
>
>
>
>
> --
> Gayu
Re: Flink Data Stream Union
No, it doesn't i even tried removing the filter and return all the values
as is received from the port.
My doubt is, is there anything system or CPU specific that fails to attach
the data to the datastream created.
On Wed, Oct 21, 2015 at 10:39 AM, Aljoscha Krettek <
aljoscha.kret...@gmail.com> wrote:
> So does the filter maybe filter out everything?
> > On 21 Oct 2015, at 16:18, Gayu wrote:
> >
> > Yes, exactly.
> >
> > On Wed, Oct 21, 2015 at 10:17 AM, Aljoscha Krettek <
> aljoscha.kret...@gmail.com> wrote:
> > So it is received in the filter but the print afterwards does not print?
> > > On 21 Oct 2015, at 15:49, Gayu wrote:
> > >
> > > The data does arrive in the second port and i am able to see that in
> the filter class received.
> > > It happens only on specific machine on which i run the code.
> > >
> > >
> > > Yes, i did forget to post here, but my program calls the
> unionMessageStreams()
> > >
> > > On Wed, Oct 21, 2015 at 9:39 AM, Aljoscha Krettek <
> aljoscha.kret...@gmail.com> wrote:
> > > Hi Gayu,
> > > could it be that no data ever arrives on the second input stream? Or
> that the filter filters out all messages?
> > >
> > > Also, in the example you posted you forgot to call
> unionMessageStreams().
> > >
> > > Cheers,
> > > Aljoscha
> > >
> > > > On 21 Oct 2015, at 15:29, Till Rohrmann
> wrote:
> > > >
> > > > Can it be that you forgot to call unionMessageStreams in your main
> method?
> > > >
> > > > Cheers,
> > > > Till
> > > >
> > > >
> > > > On Wed, Oct 21, 2015 at 3:02 PM, flinkuser wrote:
> > > > Here is the strange behavior.
> > > >
> > > > Below code works in one box but not in the other. I had it working
> in my
> > > > laptop the whole of yesterday, but strangely today it doesnt work in
> my
> > > > desktop.
> > > >
> > > > Can anyone please let me know what the issue is.
> > > >
> > > >
> > > > public static void main(String[] args) throws Exception {
> > > > try {
> > > > final StreamExecutionEnvironment env =
> > > > StreamExecutionEnvironment.getExecutionEnvironment();
> > > > DataStream msgDataStream1 =
> env.addSource((new
> > > > SocketSource(hostName1, port, '\n', -1))).filter(new
> > > > MessageFilter()).setParallelism(1);
> > > > DataStream msgDataStream2 =
> env.addSource((new
> > > > SocketSource(hostName2, port, '\n', -1))).filter(new
> > > > MessageFilter()).setParallelism(1);
> > > >
> > > > env.execute("Stock stream");
> > > >
> > > > } catch (Exception e) {
> > > > System.err.println("Exception = > " +
> e.getMessage());
> > > > e.printStackTrace();
> > > > }
> > > > }
> > > >
> > > > private static void unionMessageStreams(DataStream
> msgDataStream1,
> > > > DataStream msgDataStream2) {
> > > > try {
> > > >
> > > > DataStream ds =
> msgDataStream1.union(msgDataStream2);
> > > > ds.print();
> > > > } catch (Exception e) {
> > > > System.err.println("Exception in union
> Message Streams () = > " +
> > > > e.getMessage());
> > > > }
> > > > }
> > > >
> > > > Thanks
> > > >
> > > >
> > > >
> > > > --
> > > > View this message in context:
> http://apache-flink-user-mailing-list-archive.2336050.n4.nabble.com/Flink-Data-Stream-Union-tp3169p3196.html
> > > > Sent from the Apache Flink User Mailing List archive. mailing list
> archive at Nabble.com.
> > > >
> > >
> > >
> > >
> > >
> > > --
> > > Gayu
> >
> >
> >
> >
> > --
> > Gayu
>
>
--
Gayu
Re: Flink Data Stream Union
Hmm, that is strange.
Could you maybe send the complete program so that I could have a look?
> On 21 Oct 2015, at 16:43, Gayu wrote:
>
> No, it doesn't i even tried removing the filter and return all the values as
> is received from the port.
>
> My doubt is, is there anything system or CPU specific that fails to attach
> the data to the datastream created.
>
> On Wed, Oct 21, 2015 at 10:39 AM, Aljoscha Krettek
> wrote:
> So does the filter maybe filter out everything?
> > On 21 Oct 2015, at 16:18, Gayu wrote:
> >
> > Yes, exactly.
> >
> > On Wed, Oct 21, 2015 at 10:17 AM, Aljoscha Krettek
> > wrote:
> > So it is received in the filter but the print afterwards does not print?
> > > On 21 Oct 2015, at 15:49, Gayu wrote:
> > >
> > > The data does arrive in the second port and i am able to see that in the
> > > filter class received.
> > > It happens only on specific machine on which i run the code.
> > >
> > >
> > > Yes, i did forget to post here, but my program calls the
> > > unionMessageStreams()
> > >
> > > On Wed, Oct 21, 2015 at 9:39 AM, Aljoscha Krettek
> > > wrote:
> > > Hi Gayu,
> > > could it be that no data ever arrives on the second input stream? Or that
> > > the filter filters out all messages?
> > >
> > > Also, in the example you posted you forgot to call unionMessageStreams().
> > >
> > > Cheers,
> > > Aljoscha
> > >
> > > > On 21 Oct 2015, at 15:29, Till Rohrmann wrote:
> > > >
> > > > Can it be that you forgot to call unionMessageStreams in your main
> > > > method?
> > > >
> > > > Cheers,
> > > > Till
> > > >
> > > >
> > > > On Wed, Oct 21, 2015 at 3:02 PM, flinkuser wrote:
> > > > Here is the strange behavior.
> > > >
> > > > Below code works in one box but not in the other. I had it working in my
> > > > laptop the whole of yesterday, but strangely today it doesnt work in my
> > > > desktop.
> > > >
> > > > Can anyone please let me know what the issue is.
> > > >
> > > >
> > > > public static void main(String[] args) throws Exception {
> > > > try {
> > > > final StreamExecutionEnvironment env =
> > > > StreamExecutionEnvironment.getExecutionEnvironment();
> > > > DataStream msgDataStream1 =
> > > > env.addSource((new
> > > > SocketSource(hostName1, port, '\n', -1))).filter(new
> > > > MessageFilter()).setParallelism(1);
> > > > DataStream msgDataStream2 =
> > > > env.addSource((new
> > > > SocketSource(hostName2, port, '\n', -1))).filter(new
> > > > MessageFilter()).setParallelism(1);
> > > >
> > > > env.execute("Stock stream");
> > > >
> > > > } catch (Exception e) {
> > > > System.err.println("Exception = > " +
> > > > e.getMessage());
> > > > e.printStackTrace();
> > > > }
> > > > }
> > > >
> > > > private static void unionMessageStreams(DataStream
> > > > msgDataStream1,
> > > > DataStream msgDataStream2) {
> > > > try {
> > > >
> > > > DataStream ds =
> > > > msgDataStream1.union(msgDataStream2);
> > > > ds.print();
> > > > } catch (Exception e) {
> > > > System.err.println("Exception in union Message
> > > > Streams () = > " +
> > > > e.getMessage());
> > > > }
> > > > }
> > > >
> > > > Thanks
> > > >
> > > >
> > > >
> > > > --
> > > > View this message in context:
> > > > http://apache-flink-user-mailing-list-archive.2336050.n4.nabble.com/Flink-Data-Stream-Union-tp3169p3196.html
> > > > Sent from the Apache Flink User Mailing List archive. mailing list
> > > > archive at Nabble.com.
> > > >
> > >
> > >
> > >
> > >
> > > --
> > > Gayu
> >
> >
> >
> >
> > --
> > Gayu
>
>
>
>
> --
> Gayu
Re: Flink Data Stream Union
Hi,
first of all, am I correct to assume that
new SocketSource(hostName1, port, '\n', -1)
should be
new SocketTextStreamFunction(hostName1, port1, '\n', -1)
or are you using a custom built SocketSource for this?
If I replace it by SocketTextStreamFunction and execute it the example runs and
prints incoming Strings from both input sockets.
How are you executing the example? In the IDE or on a Flink cluster?
Cheers,
Aljoscha
> On 21 Oct 2015, at 15:02, flinkuser wrote:
>
> Here is the strange behavior.
>
> Below code works in one box but not in the other. I had it working in my
> laptop the whole of yesterday, but strangely today it doesnt work in my
> desktop.
>
> Can anyone please let me know what the issue is.
>
>
> public static void main(String[] args) throws Exception {
> try {
> final StreamExecutionEnvironment env =
> StreamExecutionEnvironment.getExecutionEnvironment();
> DataStream msgDataStream1 = env.addSource((new
> SocketSource(hostName1, port, '\n', -1))).filter(new
> MessageFilter()).setParallelism(1);
> DataStream msgDataStream2 = env.addSource((new
> SocketSource(hostName2, port, '\n', -1))).filter(new
> MessageFilter()).setParallelism(1);
>
> env.execute("Stock stream");
>
> } catch (Exception e) {
> System.err.println("Exception = > " + e.getMessage());
> e.printStackTrace();
> }
> }
>
> private static void unionMessageStreams(DataStream
> msgDataStream1,
> DataStream msgDataStream2) {
> try {
>
> DataStream ds =
> msgDataStream1.union(msgDataStream2);
> ds.print();
> } catch (Exception e) {
> System.err.println("Exception in union Message Streams
> () = > " +
> e.getMessage());
> }
> }
>
> Thanks
>
>
>
> --
> View this message in context:
> http://apache-flink-user-mailing-list-archive.2336050.n4.nabble.com/Flink-Data-Stream-Union-tp3169p3196.html
> Sent from the Apache Flink User Mailing List archive. mailing list archive at
> Nabble.com.
Re: Flink Data Stream Union
I think the most crucial question is still whether you are running 0.9.1 or
0.10-SNAPSHOT, because the 0.9.1 union has known issues...
If you are running 0.9.1 there is not much you can do except upgrade the
version ;-)
On Wed, Oct 21, 2015 at 5:19 PM, Aljoscha Krettek
wrote:
> Hi,
> first of all, am I correct to assume that
> new SocketSource(hostName1, port, '\n', -1)
> should be
> new SocketTextStreamFunction(hostName1, port1, '\n', -1)
>
> or are you using a custom built SocketSource for this?
>
> If I replace it by SocketTextStreamFunction and execute it the example
> runs and prints incoming Strings from both input sockets.
>
> How are you executing the example? In the IDE or on a Flink cluster?
>
> Cheers,
> Aljoscha
> > On 21 Oct 2015, at 15:02, flinkuser wrote:
> >
> > Here is the strange behavior.
> >
> > Below code works in one box but not in the other. I had it working in my
> > laptop the whole of yesterday, but strangely today it doesnt work in my
> > desktop.
> >
> > Can anyone please let me know what the issue is.
> >
> >
> > public static void main(String[] args) throws Exception {
> > try {
> > final StreamExecutionEnvironment env =
> > StreamExecutionEnvironment.getExecutionEnvironment();
> > DataStream msgDataStream1 =
> env.addSource((new
> > SocketSource(hostName1, port, '\n', -1))).filter(new
> > MessageFilter()).setParallelism(1);
> > DataStream msgDataStream2 =
> env.addSource((new
> > SocketSource(hostName2, port, '\n', -1))).filter(new
> > MessageFilter()).setParallelism(1);
> >
> > env.execute("Stock stream");
> >
> > } catch (Exception e) {
> > System.err.println("Exception = > " +
> e.getMessage());
> > e.printStackTrace();
> > }
> > }
> >
> > private static void unionMessageStreams(DataStream
> msgDataStream1,
> > DataStream msgDataStream2) {
> > try {
> >
> > DataStream ds =
> msgDataStream1.union(msgDataStream2);
> > ds.print();
> > } catch (Exception e) {
> > System.err.println("Exception in union Message
> Streams () = > " +
> > e.getMessage());
> > }
> > }
> >
> > Thanks
> >
> >
> >
> > --
> > View this message in context:
> http://apache-flink-user-mailing-list-archive.2336050.n4.nabble.com/Flink-Data-Stream-Union-tp3169p3196.html
> > Sent from the Apache Flink User Mailing List archive. mailing list
> archive at Nabble.com.
>
>
