Flink SQL INSERT 语句单表写入、多表分流、分区覆盖与 StatementSet

1. INSERT 语句是干嘛的

INSERT用于把查询结果或字面量数据写入目标表（sink 表）。在 Flink 里，执行 INSERT 会提交一个 Flink Job（流式作业通常是长期运行）。

---

2. Java 里怎么跑 INSERT：单条 executeSql vs 多条 StatementSet

2.1 单条 INSERT：executeSql 立即提交作业

executeSql()执行 INSERT 会立刻提交 Job，并返回TableResult，你可以通过它拿到 JobClient 查询状态。

TableEnvironmenttEnv=TableEnvironment.create(...);// source & sinktEnv.executeSql("CREATE TABLE Orders (`user` BIGINT, product VARCHAR, amount INT) WITH (...)");tEnv.executeSql("CREATE TABLE RubberOrders(product VARCHAR, amount INT) WITH (...)");// submit job immediatelyTableResultr1=tEnv.executeSql("INSERT INTO RubberOrders "+"SELECT product, amount FROM Orders WHERE product LIKE '%Rubber%'");System.out.println(r1.getJobClient().get().getJobStatus());

2.2 多条 INSERT：StatementSet.addInsertSql 延迟执行，一次 execute 提交

当你要把同一个源表数据分流写入多个 sink（比如 RubberOrders、GlassOrders），用StatementSet更合适：先addInsertSql()收集多条语句，最后execute()一次性提交。

tEnv.executeSql("CREATE TABLE GlassOrders(product VARCHAR, amount INT) WITH (...)");StatementSetstmtSet=tEnv.createStatementSet();stmtSet.addInsertSql("INSERT INTO RubberOrders SELECT product, amount FROM Orders WHERE product LIKE '%Rubber%'");stmtSet.addInsertSql("INSERT INTO GlassOrders SELECT product, amount FROM Orders WHERE product LIKE '%Glass%'");TableResultr2=stmtSet.execute();System.out.println(r2.getJobClient().get().getJobStatus());

注意：addInsertSql()每次只能接收一条 INSERT 语句（不要把多个 INSERT 拼一条字符串）。

3. INSERT INTO / INSERT OVERWRITE：追加 vs 覆盖

3.1 总体语法（Insert from Select）

[EXECUTE]INSERT{INTO|OVERWRITE }[catalog.][db.]table_name[PARTITIONpart_spec][column_list]select_statement

3.2 INTO：追加写入（Append）

• 不覆盖已有数据（或已有分区数据），新结果继续追加。

3.3 OVERWRITE：覆盖写入（Overwrite）

• INSERT OVERWRITE会覆盖目标表或目标分区已有数据。
• 常用于离线批处理、重跑分区、或者“以最后一次跑出来的结果为准”的场景。

4. 分区写入：静态分区 vs 动态分区

假设目标表是分区表：

CREATETABLEcountry_page_view(userSTRING,cntINT,dateSTRING,country STRING)PARTITIONEDBY(date,country)WITH(...);

4.1 写入静态分区（date/country 都固定）

INSERTINTOcountry_page_viewPARTITION(date='2019-8-30',country='China')SELECTuser,cntFROMpage_view_source;

4.2 半动态分区（date 固定、country 每行决定）

INSERTINTOcountry_page_viewPARTITION(date='2019-8-30')SELECTuser,cnt,countryFROMpage_view_source;

4.3 覆盖分区写入（静态/半动态都支持）

INSERTOVERWRITE country_page_viewPARTITION(date='2019-8-30',country='China')SELECTuser,cntFROMpage_view_source;INSERTOVERWRITE country_page_viewPARTITION(date='2019-8-30')SELECTuser,cnt,countryFROMpage_view_source;

5. EXECUTE 关键字：显式执行（语义等价）

Flink 允许在 INSERT 前面加EXECUTE，用于强调“我要执行这条语句”，但语义上等价于不加。

EXECUTEINSERTINTOcountry_page_viewPARTITION(date='2019-8-30',country='China')SELECTuser,cntFROMpage_view_source;

6. column_list：部分列写入（Partial Insert）怎么映射？

Flink 支持指定目标列列表，把 SELECT 的列按列表顺序写入指定列，未写到的列会被置为NULL（前提：该列可空）。

例：表T(a INT, b INT, c INT)：

INSERTINTOT(c,b)SELECTx,yFROMS;

含义是：

• x写入c
• y写入b
• a被置为NULL（如果a允许为 NULL）

对 connector/sink 开发者：可以通过DynamicTableSink.Context.getTargetColumns()获取用户指定的目标列，决定如何处理“部分列更新”。

7. INSERT … VALUES：直接插入字面量行

除了INSERT INTO ... SELECT ...，也可以直接写 values：

[EXECUTE]INSERT{INTO|OVERWRITE } table_nameVALUES(val1,val2,...),(val1,val2,...);

示例：

CREATETABLEstudents(name STRING,ageINT,gpaDECIMAL(3,2))WITH(...);INSERTINTOstudentsVALUES('fred flintstone',35,1.28),('barney rubble',32,2.32);

8. 一条 SQL 写多个表：EXECUTE STATEMENT SET

如果你在 SQL 层就想“一次提交多条 insert”，可以用：

EXECUTESTATEMENTSETBEGINinsert_statement;insert_statement;END;

其中insert_statement可以是INSERT ... SELECT或INSERT ... VALUES。

9. 生产实践建议（你放到博客结尾很加分）

1. 多 sink 分流优先用 StatementSet：一次提交、共享规划，写法更稳。
2. OVERWRITE 慎用：尤其是流式任务，确认 connector 对覆盖语义的支持与目标表期望行为。
3. 分区写入要区分静态/动态：静态分区适合重跑；动态分区适合实时按维度落地。
4. 部分列写入会把其它列写成 NULL：对非空列/主键列要提前约束，否则容易写入失败或产生脏数据。