Using Table & SQL API to build a stateful table which will just have 2 columns: (1) Rest_ID - will be 100 restaurants IDs. They will be randomly getting the Statuses either +1 or -1 implying a pending order count. Same rest_IDs can continue to have +1s and then -1s later (2) Status Count - as it implies will just be either 1 or -1. Data is ingested from an unbounded Kafka stream. And data looks like this -> {"rest_id": "6487f4c6fc13ae161d9008f6", "status": "-1"} {"rest_id": "6487f4c6fc13ae161d9008e5", "status": "1"}... I would like to showcase a stateful table (Apache Flink Runtime) where column 1 can have DISTINCT 100 rest_ID that is static and column 2 can be updating the status_count (sum of 1s and -1s statuses) as the data is ingested from kafka. I have written the boilerplate code below to ingest the kafka information (below)- I would like help to complete the code to create a stateful table and print out the results with changing status_count

from pyflink.table import EnvironmentSettings, TableEnvironment

from pyflink.table.expressions import *

def log_processing():

env_settings = <http://EnvironmentSettings.in|EnvironmentSettings.in>_streaming_mode()

t_env = TableEnvironment.create(env_settings)

##### specify connector and format jars

t_env.get_config().set("pipeline.jars", "file:///Users/karanbawejapro/Desktop/flink-sql-connector-kafka-1.17.0.jar")

t_env.get_config().set("table.exec.source.idle-timeout", "1000")

source_ddl = """

CREATE TABLE restaurant_live_pending_orders(

rest_id VARCHAR,

status VARCHAR

) WITH (

'connector' = 'kafka',

'topic' = 'live_order_status',

'properties.bootstrap.servers' = 'localhost:9092',

'properties.group.id' = 'rest_group',

'scan.startup.mode' = 'specific-offsets',

'scan.startup.specific-offsets' = 'partition:0,offset:0',

'json.fail-on-missing-field' = 'false',

'json.ignore-parse-errors' = 'true',

'format' = 'json'

)

"""

tbl = t_env.execute_sql(source_ddl)

tbl = t_env.from_path('restaurant_live_pending_orders')