Flink SQL INSERT 语句单表写入、多表分流、分区覆盖与 StatementSet
2026/5/25 17:18:50 网站建设 项目流程

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 ... SELECTINSERT ... VALUES

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

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

需要专业的网站建设服务?

联系我们获取免费的网站建设咨询和方案报价,让我们帮助您实现业务目标

立即咨询