Hi All, I have the below requirement for Flink batch mode. I have to backfill data for around 120 days, which involves below operations. 1. Read data for that day from s3 2. Process the data. 3. Push data to Kafka Sink. I am executing these steps in Loop from startDate to endDate. But instead of job executing in sequence for each day, Read step for all 120 days are being executed first and then 2nd Steps followed by 3rd Step. I want data to be pushed one day at a time in sequence manner. Can anyone help here Here is code snippet

ApplicationConfig config = context.getApplicationConfig();
    final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
    // Set run time env to batch
    env.setRuntimeMode(RuntimeExecutionMode.BATCH);
    // env.setParallelism(10);
    env.getConfig().setGlobalJobParameters(context.getParameterTool());
    LocalDate date = startDate;
    while (endDate.compareTo(date) >= 0) {
      StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);
      for (String eventName : EVENT_TYPES_LIST) {
        // Map and Kafka Sink
        processData(tableEnv, date, context, eventName, env);
      }
      date = date.plusDays(1);
    }

 
    env.execute(config.getServiceIdentifier() + "{" + startDate + " to " + endDate + "}");

Attaching screenshot of Execution Graph for reference, which shows it reading the data at once from s3

Flink's programming model is not imperative, you declare your dataflow using Flink's API and it decides on the best dataflow for your computation. My suggestion would be to repartition your dataset in a standard hive format and use the TableAPI to read your table, that will avoid the for loop and Flink will read things in time order. see Then you can convert back to datastream if you wish so and sink into kafka

@André Midea Jasiskis I want data to be processed in incremental date fashion only. Also, Previously I was using Table API to load data, but Table API cannot read GZ compressed files.

You can set a filter for the dates you want to process in a SQL query, then you'd get just the delta.

I think data for 120 days cannot be loaded in go using DataStream API and I have to convert back to Table API and perform the query for filter