from pyflink.table import EnvironmentSettings, TableEnvironment

# create a streaming TableEnvironment
env_settings = EnvironmentSettings.in_streaming_mode()
t_env = TableEnvironment.create(env_settings)
t_env.get_config().set("pipeline.jars",'file:////jars/flink-sql-connector-kafka-1.16.0.jar)

table_name = "source_table"
source_ddl = """
				create table %s(
					`@timestamp` bigint,
					ts as to_timestamp(from_unixtime(`@timestamp`)), 
                    type string,
					flowid int,
					watermark for ts as ts-interval '1' minute
				) WITH (
					'connector' = 'kafka',
					'property-version' = 'universal',
					'topic' = '%s',
					'properties.bootstrap.servers' = '%s',
					'properties.group.id' = '%s',
					'scan.startup.mode' = 'latest-offset',
					'format' = 'json'
				)
			"""
# register table and return it
t_env.execute_sql(source_ddl %(table_name, 'topic_1', 'localhost:9092', 'g1'))
t_source_table = t_env.from_path(table_name)

result = t_source_table\
                    .window(Tumble.over(lit(1).minutes).on(col('ts')).alias('one_minute_window'))\
                    .group_by(col('one_minute_window'), col('type'))\
                    .select(col('one_minute_window'), col('type'),col("flowid").count.alias("flowid_count"))

result.execute().print()