如何在单个PyFlink SQL作业（Flink 1.13.6版本）中执行多条INSERT INTO...SELECT语句？

首先明确回答：你完全可以在单个PyFlink作业中运行多条INSERT INTO ... SELECT ...语句，但你的写法有问题——每次调用env.execute_sql()都会触发Flink环境的执行操作，而一个Flink环境实例只能执行一次作业，这就是你遇到Cannot have more than one execute() or executeAsync() call in a single environment报错的核心原因。

你看到Web UI里出现第一个insert作业，是因为第一次调用execute_sql(sql1)时，Flink已经把这个查询打包成作业提交到集群并启动了，第二次调用时环境已经处于活跃执行状态，自然无法再提交第二个作业。

无需脱离SQL/Table API的解决方案 #

最直接且高效的方式是把多条INSERT语句合并成一个SQL脚本，用分号分隔，然后一次性调用execute_sql()提交：

# 合并多条INSERT语句为一个SQL脚本
combined_sql = """
INSERT INTO out1 (col1, col2) SELECT col1, col2 FROM input;
INSERT INTO out2 (col3, col4) SELECT col3, col4 FROM input;
"""
# 一次性执行合并后的SQL
result = env.execute_sql(combined_sql)
# 如果是批处理场景，需要等待作业完成可以调用wait()；流处理场景可省略
result.wait()

这样做的好处不止是解决报错：Flink的优化器会自动识别这些查询共享同一个input源，会在作业中只读取一次input的数据，然后分流写入到不同的输出表，避免重复读取带来的资源浪费，执行效率更高。

另外补充一点：如果你更倾向于用Table API的方式，也可以先把每个查询转换成Table对象，再分别调用execute_insert()方法，但要注意不要触发多次环境执行——正确的做法是先定义好所有的sink操作，最后统一调用env.execute()（不过这种方式不如合并SQL简洁，推荐前者）。

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