Calling external script from Flink

Question

This might seem like a very complex problem to some of you. I want to use Apache Flink to apply some algorithms on data from a SocketStream. However, these algorithms are external executables that I am running using Scala's sys.process package. Here is what I want Flink to do:

Get individual lines from SocketStream:

val text = env.socketTextStream(hostName, port) val lines = text.flatMap { _.toLowerCase.split("\\n") filter { _.nonEmpty } }
Call my executable algorithm with these lines as command line parameters. Somewhat like this:

var op = "./Somefile.py "+lines!
Print the output I get from the executable.

op.print()

Obviously this is not the correct way to do what I am trying to do as op unlike lines is not a data sink and thus nothing is getting printed. Is there some way I can achieve this?

Fabian Hueske · Accepted Answer · 2016-03-16 08:09:51Z

2

If you put all arguments into a single String value, you can call the external executable from a MapFunction.

This would look like:

val args: DataStream[String] = env.socketTextStream(hostName, port) 
// assume each text line has all elements
val out: DataStream[String] = args.map(new ExternalCaller())
// print result
out.print()

with

class ExternalCaller extends MapFunction[String, String] {

  override def map(args: String): String = {
    // call external executable with args here and return output
  }
}

answered Mar 16, 2016 at 8:09

Fabian Hueske

19k2 gold badges46 silver badges49 bronze badges

Sign up to request clarification or add additional context in comments.

1 Comment

Piyush Shrivastava Over a year ago

I kind of figured that it could be done with map, but this solved my problem. Thanks.

Collectives™ on Stack Overflow

Calling external script from Flink

1 Answer 1

1 Comment

Your Answer

Hot Network Questions

Collectives™ on Stack Overflow

1 Answer 1

1 Comment

Your Answer

Sign up or log in

Post as a guest

Related