Hello everyone, I’m trying to do a fairly simple task with bounded batch stream of data, but since every book I have found is on unbounded infinite streams I did not manage to find how to do it. What I’m trying to do is basically make it so that stream2 is executed only once stream1 is finished:

import org.apache.flink.streaming.api.scala._

object FlinkTestExample {
  def main(args: Array[String]): Unit = {
    val env = StreamExecutionEnvironment.getExecutionEnvironment

    val range1 = Range.inclusive(1, 10)
    val range2 = Range.inclusive(100, 120)

    val stream1 = env.fromCollection(range1)
    val stream2 = env.fromCollection(range2)
    
    stream1.map(x => {
      val res = x * 2
      println(s"stream1: $res")
      res
    })

    stream2.map(x => {
      val res = x * 1000
      println(s"stream2: $res")
      res
    })
    
    env.execute("Test Example")
  }
}

What is the simplest way to implement some kind of synchronisation mechanism for this? Thank you in advice for your time.

You could do this:

public class DataStreamJob {

   public static void main(String[] args) throws Exception {
      final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

      DataStreamSource<Long> stream1 = env.fromSequence(1, 10);
      stream1.map(x -> {
            Long res = x * 2;
            System.out.println("stream1: " + res);
            return res;
      });
      env.execute();

      DataStreamSource<Long> stream2 = env.fromSequence(100, 110);
      stream2.map(x -> {
         Long res = x * 1000;
         System.out.println("stream2: " + res);
         return res;
      });
      env.execute();

   }
}

I think I dig too deep. This works. Thank you very much, although I have one more question. Is this solution guaranteed to run in sequence one after another?

Yes. env.execute waits for the first job to complete. You would use executeAsync if you wanted to run these jobs concurrently.