Need to understand that the below code works on an event by event stream basis or in microbatch fashion.. Source: Kafka (input json data with M keys) Destination: Kafka (flattened json data with few keys ) Flink job purpose: simple transformation Psuedo code:

StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);

// ingest a DataStream from an external source
DataStream<Tuple3<Long, String, Integer>> ds = env.addSource(.. some Kafka source.);

// SQL query with an inlined (unregistered) table
Table table = tableEnv.fromDataStream(ds, $("user"), $("product"), $("amount"));
Table result = tableEnv.sqlQuery(
    "SELECT user, product, amount FROM " + table + " WHERE product LIKE '%Rubber%'");

// SQL