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：对非空列/主键列要提前约束，否则容易写入失败或产生脏数据。