flink-user mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From "Matthias J. Sax" <mj...@apache.org>
Subject Re: Flink Stream: How to ship results through socket server
Date Tue, 19 Jan 2016 19:27:39 GMT
Yes (if I understand correctly what you aim for).

On 01/19/2016 05:57 PM, Saiph Kappa wrote:
> Thanks for your reply Mattias. So it is not possible to open a socket
> server in the JobGraph and having it open during the lifetime of the
> job, is that what you are saying? And it is required to have an external
> process to open that socket server.
> 
> On Tue, Jan 19, 2016 at 5:38 PM, Matthias J. Sax <mjsax@apache.org
> <mailto:mjsax@apache.org>> wrote:
> 
>     Your "SocketWriter-Thread" code will run on your client. All code in
>     "main" runs on the client.
> 
>     execute() itself runs on the client, too. Of course, it triggers the job
>     submission to the cluster. In this step, the assembled job from the
>     previous calls is translated into the JobGraph which is submitted to the
>     JobManager for execution.
> 
>     You should start your SocketWriter-Thread manually on the cluster, ie,
>     if you use "localhost" in "env.socketTextStream", it must be the
>     TaskManager machine that executes this SocketStream-source task.
> 
>     I guess, it would be better not to use "localhost", but start your
>     SocketWriter-Thread on a dedicated machine in the cluster, and connect
>     your SocketStream-source to this machine via its host name.
> 
>     -Matthias
> 
> 
> 
>     On 01/19/2016 03:57 PM, Saiph Kappa wrote:
>     > Hi,
>     >
>     > This is a simple example that I found using Flink Stream. I changed it
>     > so the flink client can be executed on a remote cluster, and so
>     that it
>     > can open a socket server to ship its results for any other consumer
>     > machine. It seems to me that the socket server is not being open
>     in the
>     > remote cluster, but rather in my local machine (which I'm using to
>     > launch the app). How can I achieve that? I want to be able to ship
>     > results directly from the remote cluster, and through a socket server
>     > where clients can use as a tap.
>     >
>     > Sorry about indentation:
>     >
>     > |def main(args: Array[String]) { |
>     >
>     >     val env =
>     > StreamExecutionEnvironment.createRemoteEnvironment("myhostname",
>     > DefaultFlinkMasterPort,
>     >
>     > ||"myapp-assembly-0.1-SNAPSHOT.jar"); | //Read from a socket stream at
>     > map it to StockPrice objects val socketStockStream =
>     > env.socketTextStream("localhost", 9999).map(x => { val split =
>     > x.split(",") StockPrice(split(0), split(1).toDouble) }) //Generate
>     other
>     > stock streams val SPX_Stream =
>     env.addSource(generateStock("SPX")(10) _)
>     > val FTSE_Stream = env.addSource(generateStock("FTSE")(20) _) val
>     > DJI_Stream = env.addSource(generateStock("DJI")(30) _) val
>     BUX_Stream =
>     > env.addSource(generateStock("BUX")(40) _) //Merge all stock streams
>     > together val stockStream = socketStockStream.merge(SPX_Stream,
>     > FTSE_Stream, DJI_Stream, BUX_Stream) stockStream.print()
>     > |
>     >
>     > // WHERE IS THE FOLLOWING CODE RUN?
>     >
>     > |var out: PrintWriter = null
>     > new Thread {
>     > override def run(): Unit = {
>     > val serverSocket = new ServerSocket(12345)
>     > while (true) {
>     > val socket = serverSocket.accept()
>     > val hostname = socket.getInetAddress.getHostName.split('.').head
>     > println(s"Got a new connection from $hostname")
>     > out = new PrintWriter(socket.getOutputStream)
>     > }
>     > }
>     > }.start()
>     >
>     > |||stockStream|.addSink(record => {
>     > if(out != null) {
>     > out.write(record)
>     > out.flush()
>     > }
>     > })
>     >
>     > env.execute("Stock stream") }|
>     >
>     > Thanks.
> 
> 


Mime
View raw message