Flink Table API与SQL：流批一体查询

CREATE TABLE orders (  order_id STRING,  dish_name STRING,  amount DECIMAL(10, 2),  order_time TIMESTAMP(3),  WATERMARK FOR order_time AS order_time - INTERVAL '5' SECOND) WITH (  'connector' = 'kafka',  'topic' = 'orders',  'properties.bootstrap.servers' = 'localhost:9092',  'format' = 'json');

CREATE TABLE page_views (  user_id BIGINT,  page_id BIGINT,  view_time TIMESTAMP(3),  WATERMARK FOR view_time AS view_time - INTERVAL '5' SECOND) WITH (  'connector' = 'kafka',  'topic' = 'page_views',  'properties.bootstrap.servers' = 'localhost:9092',  'properties.group.id' = 'flink-group',  'scan.startup.mode' = 'earliest-offset',  'format' = 'csv',  'csv.ignore-parse-errors' = 'true');

SELECT dish_name, COUNT(*) AS cnt, SUM(amount) AS totalFROM ordersGROUP BY dish_name, TUMBLE(order_time, INTERVAL '1' HOUR);

// 流处理环境StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env); // 批处理环境ExecutionEnvironment batchEnv = ExecutionEnvironment.getExecutionEnvironment();BatchTableEnvironment batchTableEnv = BatchTableEnvironment.create(batchEnv);

WATERMARK FOR order_time AS order_time - INTERVAL '5' SECOND

   SELECT dish_name, COUNT(*) FROM orders   GROUP BY TUMBLE(order_time, INTERVAL '5' MINUTE);

   SELECT dish_name, COUNT(*) FROM orders   GROUP BY HOP(order_time, INTERVAL '5' MINUTE, INTERVAL '10' MINUTE);

   SELECT user_id, COUNT(*) FROM clicks   GROUP BY SESSION(event_time, INTERVAL '10' MINUTE);

-- 状态 TTL 配置table.exec.state.ttl=1h

原始SQL   │   ▼Calcite 逻辑计划 ──► 逻辑优化（谓词下推等）   │   ▼Flink 物理计划 ──► 物理优化（分区裁剪等）   │   ▼StreamGraph / ExecGraph

import timeimport randomfrom datetime import datetime, timedelta class FlinkTableDemo:    """模拟 Flink Table API 流批一体查询的核心概念"""     def __init__(self):        # 模拟 TableEnvironment：注册表和连接器        self.tables = {}        self.connectors = {}        self.watermark_delay = 5  # 允许事件迟到 5 秒            def register_table(self, name, schema, connector):        """注册表：声明表结构 + 连接器配置"""        print(f"[TableEnv] 注册表 '{name}': schema={schema}, connector={connector}")        self.tables[name] = {"schema": schema, "rows": [], "connector": connector}        return True     def create_connector(self, name, source_type, config):        """创建连接器：模拟 Kafka/文件等数据源连接"""        print(f"[Connector] 创建连接器 '{name}': type={source_type}, config={config}")        self.connectors[name] = {"type": source_type, "config": config}        return name     def ingest_stream(self, table_name, events):        """模拟流式数据接入：事件不断到达"""        for event in events:            event["_arrival_time"] = datetime.now()            # 事件时间 = 业务时间（模拟订单时间）            event["_event_time"] = event["ts"]        self.tables[table_name]["rows"].extend(events)        return len(events)     def add_watermark(self, table_name):        """计算当前水位线：最新事件时间 - 延迟"""        rows = self.tables[table_name]["rows"]        if not rows:            return None        max_event_time = max(r["_event_time"] for r in rows)        watermark = max_event_time - timedelta(seconds=self.watermark_delay)        return watermark     def tumbling_window_agg(self, table_name, window_sec):        """滚动窗口聚合：按固定窗口大小统计"""        rows = self.tables[table_name]["rows"]        result = {}        for r in rows:            et = r["_event_time"]            # 计算窗口起始时间：把事件时间对齐到窗口边界            ts = int(et.timestamp())            win_start = (ts // window_sec) * window_sec            win_key = datetime.fromtimestamp(win_start)            if win_key not in result:                result[win_key] = {"count": 0, "total": 0.0, "items": []}            result[win_key]["count"] += 1            result[win_key]["total"] += r["amount"]            result[win_key]["items"].append(r["dish"])        return result     def continuous_query(self, table_name, agg_func):        """模拟连续查询：返回查询函数模拟持续监控"""        def query():            watermark = self.add_watermark(table_name)            rows = self.tables[table_name]["rows"]            # 只处理水位线之前的数据（模拟 Flink 的触发机制）            valid = [r for r in rows if r["_event_time"] <= watermark] if watermark else []            return agg_func(valid)        return query     def dynamic_table_to_stream(self, agg_result, mode="retract"):        """动态表转流：模拟三种输出模式"""        stream = []        if mode == "append":            for win, data in agg_result.items():                stream.append(("INSERT", {"window": str(win), "count": data["count"]}))        elif mode == "retract":            for win, data in agg_result.items():                stream.append(("INSERT", {"window": str(win), "count": data["count"]}))                # Retract 模式下，更新时先发 DELETE 再发 INSERT                stream.append(("DELETE", {"window": str(win), "count": data["count"]}))                stream.append(("INSERT", {"window": str(win), "count": data["count"]}))        elif mode == "upsert":            seen = {}            for win, data in agg_result.items():                key = str(win)                if key in seen:                    stream.append(("UPDATE", {"key": key, "count": data["count"]}))                else:                    stream.append(("INSERT", {"key": key, "count": data["count"]}))                seen[key] = data["count"]        return stream  if __name__ == "__main__":    # ========== 模拟完整 Flink 任务流程 ==========        # 1. 创建 TableEnvironment    env = FlinkTableDemo()    print("=== 1. 创建 TableEnvironment ===")        # 2. 创建连接器（对接 Kafka）    kafka_connector = env.create_connector(        "kafka-orders",        "kafka",        {"bootstrap.servers": "localhost:9092", "topic": "orders"}    )        # 3. 注册表（定义订单表结构）    env.register_table(        "orders",        schema={            "order_id": "STRING",            "dish": "STRING",            "amount": "DECIMAL(10,2)",            "ts": "TIMESTAMP(3)"        },        connector=kafka_connector    )        # 4. 接入流数据（模拟订单事件到达）    print("\n=== 2. 流式接入订单数据 ===")    now = datetime.now()    sample_orders = [        {"order_id": "001", "dish": "宫保鸡丁", "amount": 42.00,         "ts": now - timedelta(seconds=3)},        {"order_id": "002", "dish": "麻婆豆腐", "amount": 35.50,         "ts": now - timedelta(seconds=2)},        {"order_id": "003", "dish": "宫保鸡丁", "amount": 42.00,         "ts": now - timedelta(seconds=1)},        {"order_id": "004", "dish": "回锅肉", "amount": 48.00,         "ts": now},        # 一条迟到 8 秒的事件（超过 watermark_delay=5 秒）        {"order_id": "005", "dish": "宫保鸡丁", "amount": 42.00,         "ts": now - timedelta(seconds=8)},    ]    env.ingest_stream("orders", sample_orders)    print(f"已接入 {len(sample_orders)} 条订单")        # 5. 查看当前水位线    watermark = env.add_watermark("orders")    print(f"\n=== 3. 当前水位线 ===")    print(f"最新事件时间: {max(r['ts'] for r in sample_orders)}")    print(f"水位线 (= 最新事件时间 - {env.watermark_delay}s): {watermark}")        # 6. 执行连续查询（滚动窗口聚合）    print("\n=== 4. 连续查询：5秒滚动窗口聚合 ===")    continuous_query = env.continuous_query("orders", lambda rows: rows)        # 7. 执行具体聚合（模拟窗口触发）    @staticmethod    def count_dish_agg(rows):        stats = {}        for r in rows:            dish = r["dish"]            if dish not in stats:                stats[dish] = {"count": 0, "total": 0.0}            stats[dish]["count"] += 1            stats[dish]["total"] += r["amount"]        return stats        # 手动触发一次查询（模拟窗口触发）    all_valid = continuous_query()    print(f"水位线之前的事件数: {len(all_valid)}")    print("各菜品统计:")    for dish, stat in count_dish_agg(all_valid):        print(f"  {dish}: 销量={stat['count']}, 销售额={stat['total']:.2f}元")        # 8. 动态表转流（Upsert 模式）    print("\n=== 5. 动态表转流（Upsert模式）===")    agg_result = env.tumbling_window_agg("orders", window_sec=5)    stream = env.dynamic_table_to_stream(agg_result, mode="upsert")    for op in stream:        print(f"  [{op[0]}] {op[1]}")