<= br>

2015-03-12 1:28 GMT+01:00 Emmanuel <eleroy@msn= .com>:

I don't see an 'open()' function to overr= ide in the RichSinkFunction or the SinkFunction... so where is this open() = function supposed to be?
Date: Thu, 12 Mar 2015 01:17:= 34 +0100
Subject: Re: Socket output stream
From: fhueske@gmail.com
To: user@flink.apache.org

Hi Emmanuel,

the open() method should the right place for setting up the socket conne= ction. It is called on the worker node before the first input arrives.

Best, Fabian

2015-03-12 1:05 = GMT+01:00 Emmanuel <eleroy@msn.com>:

Hi Marton,

Thanks for the info.

I've been trying to implement a socket sink but = running into 'Not Serializable' kind of issues.
I was see= ing in the Spark docs that this is typically an issue, where the socket sho= uld be created on the worker node, as it can't be serialized to be move= d from the supervisor.
http://spark.apache.org/docs/1.1.0/streaming-programming-= guide.html#design-patterns-for-using-foreachrdd

So, not sure how this would be implemented in Flink...
My attem= pt (maybe very naive) looked like this:

public static final class SocketSink extends = RichSinkFunction<String> {

private PrintWriter out;

public SocketSink(String host, Integer port) throws IOException {
= Socket clientSocket =3D ne= w Socket(host,port);
out =3D ne= w PrintWriter(clientSocket.getOutputStream(), true);
}

@Override
= public void invoke(St= ring s) {
out.println(s);
}
}

maybe i should just move to Kafka directly... ;/
= Thanks for help
Emmanuel

From: mbalassi@apache.org
Dat= e: Wed, 11 Mar 2015 16:37:41 +0100
Subject: Fwd: Flink questions
To: = eleroy@msn.com
C= C: rmetzger@apache= .org; hsaputra= @apache.org; user@flink.apache.org

Dear Emmanuel,

I'm Marton, one of the Flink Streaming developers = - Robert forwarded your issue to me. Thanks for trying out our project.

1) Debugging: TaskManager logs are currently not forw= arded to the UI, but you can find them on the taskmanager machines in the l= og folder of your Flink distribution. We have this issue on our agenda in t= he very near future - they need to be accessible from the UI.
2) Output to socket: Currently we do not have a preimplemented = sink for sockets (although we offer a socket source and sinks writing to Ap= ache Kafka, Flume and RabbitMQ). You can easily implement a socket sink by = extending the abstract RichSinkFunction class though. [1]

For using that you can simply say dataStream.addSink(MySinkFunction= ()) - in that you can bring up a socket or any other service. You would cre= ate a socket in the open function and then in the invoke method you would w= rite every value out to it.

I do agree that this i= s a nice tool to have so I have opened a JIRA ticket for it. [2]
=
3) Internal data format: Robert was kind enough to offer a m= ore detailed answer on this issue. In general streaming sinks support any f= ile output that is supported by batch Flink including Avro. You can use thi= s functionality by dataStream.addSink(new FileSinkFunction<>(OutputFo= rmat)).

[1] http://ci.apache.org/projects/flink/flink-docs-mast= er/streaming_guide.html#connecting-to-the-outside-world
[= 2]=C2=A0https://issues.apache.org/jira/browse/FLINK-1688
=
Best,

Marton
From: Emmanuel <eleroy@msn.com>
Date: 11. M=C3= =A4rz 2015 14:59:31 MEZ
To: Robert Metzger <rmetzger@apache.org>, Henry S= aputra <hsaputr= a@apache.org>
Subject: Flink questions

Hello,

Thanks aga= in for the help yesterday: the simple things go a long way to get me moving= ...

I have more questions i hope I can get your op= inion and input about:

Debugging:
What's the preferred or recommended way to proceed?=C2=A0
I = have been using some System.out.println() statements in my simple test code= , and the results are confusing:
First, in the UI, the logs are f= or the jobmanager.out, but there is never anything there; wherever i see ou= tput in a log it's on the taskmanager.out file
Also, even mor= e confusing is the fact that often times I just get no log at all... the UI= says the topology is running, but nothing get printed out...
Is = there a process you'd recommend to follow to debug properly with logs?= =C2=A0

Output to socket
Ideally I= 'd like to print out to a socket/stream and read from another machine s= o as not to choke the node with disk I/Os when testing performances. Not su= re how to do that.

Internal Data format
Finally, a practical question about data format: we ingest JSON, whi= ch is not convenient, and uses a lot of space. Internally Java/Scala prefer= s Tuples, and we were thinking of using ProtoBuffs.=C2=A0
There i= s also Avro that could do this as I understand it... What would be the reco= mmended way to format data internally?

Thanks for = your input.

Cheers
Emmanuel
=