如何在Flink SQL CLI中保存含NOW()的查询结果至本地文件？

问题原因 #

你之前的INSERT INTO任务无法完成，是因为如果tableOne是流表，Flink会默认以流处理模式运行查询，持续等待新数据流入，导致任务一直处于运行状态。要保留NOW()计算时的固定时间戳并完成任务，需要切换到批处理模式处理全量数据，或针对流表执行一次性快照查询。

具体实现方法 #

方法一：批处理模式导出到本地文件 #

1. 启动批处理模式的Flink SQL CLI
   执行以下命令启动CLI并指定批处理运行时模式：

   # Linux/macOS
   ./sql-client.sh -m local -e "execution.runtime-mode=batch"
   # Windows
   sql-client.bat -m local -e "execution.runtime-mode=batch"
   

2. 创建本地文件输出表
   在CLI中创建用于存储结果的文件表，替换路径和字段为你的实际需求：

   CREATE TABLE local_result (
       -- 定义与查询结果匹配的字段，包含NOW()生成的时间字段
       id INT,
       content STRING,
       query_time TIMESTAMP(3)
   ) WITH (
       'connector' = 'filesystem',
       'path' = 'file:///home/yourname/flink_result', -- 本地文件路径，Windows用file:///C:/xxx/result
       'format' = 'csv' -- 支持csv、json、parquet等格式
   );
   

3. 执行插入任务
   执行INSERT语句，此时Flink会以批处理方式处理tableOne的全量数据，NOW()会在查询启动时一次性计算出固定时间戳，任务处理完所有数据后自动终止：

   INSERT INTO local_result
   SELECT id, content, NOW() AS query_time
   FROM tableOne;
   

方法二：流表快照查询导出 #

如果tableOne是流表且无法切换批处理模式，可以通过指定启动模式获取全量快照后导出：

1. 先创建上述的local_result表
2. 执行带启动模式的插入语句，获取流表的全量数据快照：

   INSERT INTO local_result
   SELECT id, content, NOW() AS query_time
   FROM tableOne
   /*+ OPTIONS('scan.startup.mode'='earliest-offset') */;
   

   待任务完成全量数据处理后，手动在Flink Web UI或CLI中停止任务即可，此时NOW()的时间戳为任务启动时的固定值。

内容的提问来源于stack exchange，提问作者lanny