在flink代码中写flink sql怎么比较美观

### 编写美观的Flink SQL最佳实践

#### 1. 统一编码风格
为了确保SQL语句易于阅读和维护，建议采用统一的编码风格。这包括合理使用缩进、空格以及换行来分隔不同的逻辑部分。

SELECT 
    user_id, 
    COUNT(*) AS click_count,
    SUM(amount) AS total_amount
FROM 
    clicks
WHERE 
    event_time >= '2023-01-01' AND 
    event_time < '2023-02-01'
GROUP BY 
    user_id;

#### 2. 参数化查询
通过参数化可以提高安全性并减少硬编码带来的错误风险。对于动态输入的部分应尽可能利用外部配置文件或程序变量传递给SQL模板[^1]。

// Java代码片段设置表环境
TableEnvironment tableEnv = ...;

// 定义时间范围作为参数
String startTimeParam = "startTime";
String endTimeParam = "endTime";

tableEnv.executeSql(
    String.format("""
        SELECT * FROM sales_data 
        WHERE order_date BETWEEN $%s$ AND $%s$
        """, startTimeParam, endTimeParam));

#### 3. 注释说明
适当添加注释有助于他人理解复杂业务逻辑背后的意图。特别是当涉及多个子查询或者窗口函数时更应该如此做[^2]。

-- 计算每日活跃用户的数量，并按日期降序排列
WITH daily_active_users AS (
    /* 子查询用于获取每天登录系统的唯一用户 */
    SELECT DATE(event_timestamp) as day, COUNT(DISTINCT user_id) as dau
    FROM login_events
    GROUP BY DATE(event_timestamp)
)

SELECT *
FROM daily_active_users
ORDER BY day DESC;

#### 4. 利用CTE简化表达
公共表表达式（Common Table Expressions, CTE）可以使复杂的多层嵌套变得直观易懂。它允许我们将中间结果集命名以便后续重用，在处理层次关系数据方面尤为有用。

WITH product_sales AS (
    -- 获取每种商品在过去一个月内的销售额
    SELECT p.product_name, SUM(s.quantity * s.price_per_unit) as revenue
    FROM products p JOIN sales s ON p.id = s.product_id
    WHERE sale_date >= CURRENT_DATE - INTERVAL '1 month'
    GROUP BY p.product_name
),

top_selling_products AS (
    -- 找出销售金额最高的前五名产品
    SELECT ps.*, ROW_NUMBER() OVER (ORDER BY ps.revenue DESC) rn
    FROM product_sales ps
)

SELECT tsp.*
FROM top_selling_products tsp
WHERE tsp.rn <= 5;

#### 5. 结合API特性优化性能
充分利用`Flink Table API`提供的各种内置函数与聚合操作符能够有效提升执行效率。例如应用增量计算模式代替全量扫描；或是借助物化视图缓存频繁访问的结果集等技术手段。

val tEnv: StreamTableEnvironment = ...

tEnv.createTemporaryView("clicks", ...)
tEnv.createTemporaryView("users", ...)

// 使用滑动窗口统计最近十分钟内各城市的点击次数
val result = tEnv.sqlQuery("""
    SELECT city, TUMBLE_START(watermark, INTERVAL '10' MINUTE), count(*)
    FROM clicks
    WINDOW tumble over 10 minutes on watermark as w
    GROUP BY city, w
""")

result.toRetractStream[(String, Long)].print()