Oceanus 创建

from pyflink.datastream import StreamExecutionEnvironment
from pyflink.table import TableEnvironment, DataTypes
from pyflink.table.udf import udf

# 创建执行环境
env = StreamExecutionEnvironment.get_execution_environment()
t_env = TableEnvironment.create(env)

# 定义数据源
source_ddl = """
    CREATE TABLE my_source (
        id INT,
        value DOUBLE,
        event_time TIMESTAMP(3)
    ) WITH (
        'connector' = 'kafka',
        'topic' = 'my_topic',
        'properties.bootstrap.servers' = 'localhost:9092',
        'format' = 'json'
    )
"""
t_env.execute_sql(source_ddl)

# 定义UDF计算平均值
@udf(input_types=[DataTypes.DOUBLE()], result_type=DataTypes.DOUBLE())
def calculate_average(value):
    return value

# 注册UDF并创建计算表
t_env.register_function("calculate_average", calculate_average)
result_ddl = """
    CREATE TABLE result_table (
        window_start TIMESTAMP(3),
        window_end TIMESTAMP(3),
        avg_value DOUBLE
    ) WITH (
        'connector' = 'print'
    )
"""
t_env.execute_sql(result_ddl)

# 执行查询
query = """
    SELECT 
        TUMBLE_START(event_time, INTERVAL '1' MINUTE) AS window_start,
        TUMBLE_END(event_time, INTERVAL '1' MINUTE) AS window_end,
        calculate_average(value) AS avg_value
    FROM my_source
    GROUP BY TUMBLE(event_time, INTERVAL '1' MINUTE)
"""
t_env.execute_sql(query)