Flink SQL 查询执行全解：从 sqlQuery()、executeSql() 到 TableResult 的源码级指南

大数据流处理批处理数据工程【免费下载链接】flink项目地址https://gitcode.com/gh_mirrors/fli/flink点击查看免费下载本文以 Flink 官方文档《Queries》为核心系统讲解 Flink Table API 与 SQL 混合查询的完整生命周期如何用TableEnvironment.sqlQuery()定义查询、如何用executeSql()与Table.execute()执行查询、TableResult.collect()/print()在不同 checkpointing 模式下的交付语义以及 Flink SQL 支持的 BNF 语法超集与标识符规则。结合TableEnvironmentImpl、TableImpl等源码实现读者可以准确掌握从编写 SQL 字符串到结果数据回流客户端的每一步底层机制。查询的两种入口sqlQuery() 与 executeSql()SELECT和VALUES语句通过TableEnvironment的sqlQuery()方法指定。该方法把查询结果以Table对象返回Table可以参与后续 SQL 和 Table API 混合查询混合查询会被整体优化并翻译为单一程序转换为 DataStream 继续做流式处理写入TableSink落地存储。而executeSql()则用于真正执行 SELECT / VALUES 语句将内容收集到本地客户端返回类型为TableResult。两者在 TableEnvironmentImpl 中的实现清晰体现了这一分工// flink-table/flink-table-api-java/.../TableEnvironmentImpl.java public Table sqlQuery(String query) { ListOperation operations getParser().parse(query); if (operations.size() ! 1) { throw new ValidationException( Unsupported SQL query! sqlQuery() only accepts a single SQL query.); } Operation operation operations.get(0); if (operation instanceof QueryOperation !(operation instanceof ModifyOperation)) { return createTable((QueryOperation) operation); } else { throw new ValidationException( Unsupported SQL query! sqlQuery() only accepts a single SQL query of type SELECT, UNION, INTERSECT, EXCEPT, VALUES, and ORDER_BY.); } } public TableResult executeSql(String statement) { ListOperation operations getParser().parse(statement); if (operations.size() ! 1) { throw new TableException(UNSUPPORTED_QUERY_IN_EXECUTE_SQL_MSG); } return executeInternal(operations.get(0)); }从源码结构看两条关键约束可直接指导实践两个方法都只接受单条 SQL 语句——解析出的Operation数量不等于 1 时立即抛出ValidationException/TableException。多条语句需要走createStatementSet()或INSERT INTO等 DML 通道。sqlQuery()只接受 DQL 类语句SELECT、UNION、INTERSECT、EXCEPT、VALUES、ORDER_BY即QueryOperation且非ModifyOperationDML 语句INSERT、CREATE TABLE 等必须通过executeSql()执行。不支持的 SQL 特性会导致TableException各特性在批/流下的支持范围见后文 Operations 一节。在 SQL 中引用表注册方式与内联表SQL 查询访问表的前提是表已在TableEnvironment中注册。可用的注册方式包括从 TableSource 注册从Table对象注册通过CREATE TABLE语句注册从 DataStream 注册fromDataStream/createTemporaryView或直接注册 Catalog 以指定数据源位置参见 Catalog 文档 对应的官方说明 dev/table/catalogs。Table.toString()内联表的自动注册机制一个容易忽视但非常实用的特性是Table.toString()会自动把该表在TableEnvironment下注册为一个唯一的临时视图名并返回该名称因此Table对象可以直接内联进 SQL 字符串。其实现见 TableImpl#toStringOverride public String toString() { if (tableName null) { tableName UnnamedTable$ uniqueId.getAndIncrement(); tableEnvironment.createTemporaryView(tableName, this); } return tableName; }首次调用toString()时生成UnnamedTable$id形式的临时视图名并注册后续调用复用同一名称。这也解释了为什么 Java 代码中可以用字符串拼接把Table写进 FROM 子句。指定查询Java / Scala / Python 三语言完整示例以下示例演示在注册表与内联表上指定 SQL 查询并展示如何将查询结果通过 INSERT 写入TableSink文件系统 connector CSV 格式。JavaStreamExecutionEnvironment env StreamExecutionEnvironment.getExecutionEnvironment(); StreamTableEnvironment tableEnv StreamTableEnvironment.create(env); // 从外部源接入一个 DataStream DataStreamTuple3Long, String, Integer ds env.addSource(...); // SQL 查询内联未注册的表 Table table tableEnv.fromDataStream(ds, $(user), $(product), $(amount)); Table result tableEnv.sqlQuery( SELECT SUM(amount) FROM table WHERE product LIKE %Rubber%); // SQL 查询注册表 // 将 DataStream 注册为视图 Orders tableEnv.createTemporaryView(Orders, ds, $(user), $(product), $(amount)); // 在表上执行 SQL 查询以新的 Table 获取结果 Table result2 tableEnv.sqlQuery( SELECT product, amount FROM Orders WHERE product LIKE %Rubber%); // 创建并注册 TableSink final Schema schema Schema.newBuilder() .column(product, DataTypes.STRING()) .column(amount, DataTypes.INT()) .build(); final TableDescriptor sinkDescriptor TableDescriptor.forConnector(filesystem) .schema(schema) .option(path, /path/to/file) .format(FormatDescriptor.forFormat(csv) .option(field-delimiter, ,) .build()) .build(); tableEnv.createTemporaryTable(RubberOrders, sinkDescriptor); // 对表执行 INSERT SQL将结果写入 TableSink tableEnv.executeSql( INSERT INTO RubberOrders SELECT product, amount FROM Orders WHERE product LIKE %Rubber%);Scalaval env StreamExecutionEnvironment.getExecutionEnvironment val tableEnv StreamTableEnvironment.create(env) // 从外部源读取 DataStream val ds: DataStream[(Long, String, Integer)] env.addSource(...) // SQL 查询内联未注册的表 val table ds.toTable(tableEnv, $user, $product, $amount) val result tableEnv.sqlQuery( sSELECT SUM(amount) FROM $table WHERE product LIKE %Rubber%) // SQL 查询注册表将 DataStream 以 Orders 之名注册 tableEnv.createTemporaryView(Orders, ds, $user, $product, $amount) val result2 tableEnv.sqlQuery( SELECT product, amount FROM Orders WHERE product LIKE %Rubber%) // 创建并注册 TableSink val schema Schema.newBuilder() .column(product, DataTypes.STRING()) .column(amount, DataTypes.INT()) .build() val sinkDescriptor TableDescriptor.forConnector(filesystem) .schema(schema) .format(FormatDescriptor.forFormat(csv) .option(field-delimiter, ,) .build()) .build() tableEnv.createTemporaryTable(RubberOrders, sinkDescriptor) // 对表执行 INSERT SQL将结果写入 TableSink tableEnv.executeSql( INSERT INTO RubberOrders SELECT product, amount FROM Orders WHERE product LIKE %Rubber%)Pythonenv StreamExecutionEnvironment.get_execution_environment() table_env StreamTableEnvironment.create(env) # SQL 查询内联未注册的表元素类型BIGINT, STRING, BIGINT table table_env.from_elements(..., [user, product, amount]) result table_env \ .sql_query(SELECT SUM(amount) FROM %s WHERE product LIKE %%Rubber%% % table) # 创建并注册 TableSink schema Schema.new_builder() \ .column(product, DataTypes.STRING()) \ .column(amount, DataTypes.INT()) \ .build() sink_descriptor TableDescriptor.for_connector(filesystem) \ .schema(schema) \ .format(FormatDescriptor.for_format(csv) .option(field-delimiter, ,) .build()) \ .build() t_env.create_temporary_table(RubberOrders, sink_descriptor) # 对表执行 INSERT SQL将结果写入 TableSink table_env \ .execute_sql(INSERT INTO RubberOrders SELECT product, amount FROM Orders WHERE product LIKE %Rubber%)要点说明TableDescriptor.forConnector(filesystem)中的 connector 名对应 classpath 上通过 SPI 注册的表工厂path是 connector 专属选项CSV 格式的field-delimiter是 format 专属选项。createTemporaryTable创建的表仅存在于当前TableEnvironment的临时层不会写入底层 Catalog适合程序内一次性使用跨会话可见的表应使用createTable或CREATE TABLE语句。Python 示例中%在 LIKE 表达式里需要双写为%%这是 Python 字符串格式化规则而非 SQL 语法。执行查询TableResult 的生命周期与交付语义SELECT / VALUES 语句可以通过TableEnvironment.executeSql()执行并把内容收集到本地返回的TableResult等价于执行了一个 SELECT。类似地Table.execute()也可以执行一个Table对象并把查询内容收集到本地客户端见 TableImpl#execute。TableResult.collect()返回一个可关闭的行迭代器CloseableIteratorRow。必须注意select 作业不会自行结束直到所有结果数据被收集完毕——需要主动调用CloseableIterator#close()关闭迭代器以避免资源泄漏推荐 Java 用 try-with-resources、Scala 用 try/finally 收尾TableResult中的数据只能访问一次collect()与print()不能先后互相调用。从源码看executeSql()对 SELECT 语句的执行链路是executeInternal 将QueryOperation包装为CollectModifyOperation在作业末尾追加一个 CollectSink经 planner 翻译为Transformation列表后由 executeQueryOperation 异步提交Pipeline到ExecutionEnvironment并把JobClient与ResultProvider绑定后构建TableResultImpl。这正是结果只能消费一次、作业要等数据收完才结束的底层原因。不同 Checkpointing 模式下的交付保证TableResult.collect()与TableResult.print()的行为随 checkpointing 配置不同而不同开启流作业 checkpointing 的配置参见 checkpointing config场景交付语义说明批作业或无 checkpointing 的流作业无 exactly-once / at-least-once 保证结果一旦产出即可被客户端访问但作业失败重启时会抛异常开启 exactly-once checkpointing 的流作业端到端 exactly-once结果仅在对应 checkpoint 完成后才对客户端可见开启 at-least-once checkpointing 的流作业端到端 at-least-once结果一旦产出即可访问但同一条结果可能被投递多次执行查询示例StreamExecutionEnvironment env StreamExecutionEnvironment.getExecutionEnvironment(); StreamTableEnvironment tableEnv StreamTableEnvironment.create(env, settings); tableEnv.executeSql(CREATE TABLE Orders (user BIGINT, product STRING, amount INT) WITH (...)); // 执行 SELECT 语句 TableResult tableResult1 tableEnv.executeSql(SELECT * FROM Orders); // 用 try-with-resources 保证迭代器自动关闭 try (CloseableIteratorRow it tableResult1.collect()) { while (it.hasNext()) { Row row it.next(); // 处理 row } } // 执行 Table TableResult tableResult2 tableEnv.sqlQuery(SELECT * FROM Orders).execute(); tableResult2.print();val env StreamExecutionEnvironment.getExecutionEnvironment() val tableEnv StreamTableEnvironment.create(env, settings) // 开启 checkpointing tableEnv.getConfig.set( CheckpointingOptions.CHECKPOINTING_MODE, CheckpointingMode.EXACTLY_ONCE) tableEnv.getConfig.set( CheckpointingOptions.CHECKPOINTING_INTERVAL, Duration.ofSeconds(10)) tableEnv.executeSql(CREATE TABLE Orders (user BIGINT, product STRING, amount INT) WITH (...)) // 执行 SELECT 语句 val tableResult1 tableEnv.executeSql(SELECT * FROM Orders) val it tableResult1.collect() try while (it.hasNext) { val row it.next // 处理 row } finally it.close() // 关闭迭代器以避免资源泄漏 // 执行 Table val tableResult2 tableEnv.sqlQuery(SELECT * FROM Orders).execute() tableResult2.print()env StreamExecutionEnvironment.get_execution_environment() table_env StreamTableEnvironment.create(env, settings) # 开启 checkpointing table_env.get_config().set(execution.checkpointing.mode, EXACTLY_ONCE) table_env.get_config().set(execution.checkpointing.interval, 10s) table_env.execute_sql(CREATE TABLE Orders (user BIGINT, product STRING, amount INT) WITH (...)) # 执行 SELECT 语句 table_result1 table_env.execute_sql(SELECT * FROM Orders) table_result1.print() # 执行 Table table_result2 table_env.sql_query(SELECT * FROM Orders).execute() table_result2.print()注意 Scala/Python 示例中显式开启了 exactly-once checkpointing这正是上一节语义表中结果在 checkpoint 完成后才对客户端可见那条规则的现场演示。语法BNF 超集与标识符规则Flink 使用 Apache Calcite 解析 SQL支持标准 ANSI SQL。以下 BNF 文法描述了批查询与流查询支持特性集的超集各特性在批/流下的适用性由 Operations 各节单独说明query: values | WITH withItem [ , withItem ]* query | { select | selectWithoutFrom | query UNION [ ALL ] query | query EXCEPT query | query INTERSECT query } [ ORDER BY orderItem [, orderItem ]* ] [ LIMIT { count | ALL } ] [ OFFSET start { ROW | ROWS } ] [ FETCH { FIRST | NEXT } [ count ] { ROW | ROWS } ONLY] withItem: name [ ( column [, column ]* ) ] AS ( query ) orderItem: expression [ ASC | DESC ] select: SELECT [ ALL | DISTINCT ] { * | projectItem [, projectItem ]* } FROM tableExpression [ WHERE booleanExpression ] [ GROUP BY { groupItem [, groupItem ]* } ] [ HAVING booleanExpression ] [ WINDOW windowName AS windowSpec [, windowName AS windowSpec ]* ] selectWithoutFrom: SELECT [ ALL | DISTINCT ] { * | projectItem [, projectItem ]* } projectItem: expression [ [ AS ] columnAlias ] | tableAlias . * tableExpression: tableReference [, tableReference ]* | tableExpression [ NATURAL ] [ LEFT | RIGHT | FULL ] JOIN tableExpression [ joinCondition ] joinCondition: ON booleanExpression | USING ( column [, column ]* ) tableReference: tablePrimary [ matchRecognize ] [ [ AS ] alias [ ( columnAlias [, columnAlias ]* ) ] ] tablePrimary: [ TABLE ] tablePath [ dynamicTableOptions ] [systemTimePeriod] [[AS] correlationName] | LATERAL TABLE ( functionName ( expression [, expression ]* ) ) | [ LATERAL ] ( query ) | UNNEST ( expression ) tablePath: [ [ catalogName . ] databaseName . ] tableName systemTimePeriod: FOR SYSTEM_TIME AS OF dateTimeExpression dynamicTableOptions: /* OPTIONS(keyval [, keyval]*) */ key: stringLiteral val: stringLiteral values: VALUES expression [, expression ]* groupItem: expression | ( ) | ( expression [, expression ]* ) | CUBE ( expression [, expression ]* ) | ROLLUP ( expression [, expression ]* ) | GROUPING SETS ( groupItem [, groupItem ]* ) windowRef: windowName | windowSpec windowSpec: [ windowName ] ( [ ORDER BY orderItem [, orderItem ]* ] [ PARTITION BY expression [, expression ]* ] [ RANGE numericOrIntervalExpression {PRECEDING} | ROWS numericExpression {PRECEDING} ] ) matchRecognize: MATCH_RECOGNIZE ( [ PARTITION BY expression [, expression ]* ] [ ORDER BY orderItem [, orderItem ]* ] [ MEASURES measureColumn [, measureColumn ]* ] [ ONE ROW PER MATCH ] [ AFTER MATCH ( SKIP TO NEXT ROW | SKIP PAST LAST ROW | SKIP TO FIRST variable | SKIP TO LAST variable | SKIP TO variable ) ] PATTERN ( pattern ) [ WITHIN intervalLiteral ] DEFINE variable AS condition [, variable AS condition ]* ) measureColumn: expression AS alias pattern: patternTerm [ | patternTerm ]* patternTerm: patternFactor [ patternFactor ]* patternFactor: variable [ patternQuantifier ] patternQuantifier: * | *? | | ? | ? | ?? | { { [ minRepeat ], [ maxRepeat ] } } [?] | { repeat }文法覆盖的要点查询骨架query由values、CTEWITH或 select 系列含 UNION / EXCEPT / INTERSECT组成可附加ORDER BY、LIMIT、OFFSET、FETCH FIRST/NEXT ... ROWS ONLY四个尾部子句表引用tablePath支持catalogName . databaseName . tableName三段式全限定名dynamicTableOptions/* OPTIONS(kv) */是查询内临时覆盖表 connector 选项的 hintsystemTimePeriodFOR SYSTEM_TIME AS OF ...用于时间旅行查询GROUP BY 扩展支持CUBE、ROLLUP、GROUPING SETS以及匿名分组GROUP BY ()MATCH_RECOGNIZE完整的模式匹配子句含PATTERN、DEFINE、AFTER MATCH SKIP ...、WITHIN等量词支持*、、?及{m,n}区间。标识符与字符串字面量规则Flink SQL 对标识符表名、属性名、函数名采用类似 Java 的词法策略标识符是否加引号大小写都会被保留随后按大小写敏感方式匹配与 Java 不同反引号允许标识符包含非字母数字字符例如SELECT a AS \my field FROM t。字符串字面量必须用单引号包裹如SELECT Hello World单个引号需双写转义如SELECT Its meFlink SQL SELECT Hello World, Its me; ---------------------- || EXPR$0 | EXPR$1 | ---------------------- || Hello World | Its me | ---------------------- 1 row in set字符串字面量支持 Unicode 字符若需显式指定 Unicode 码点默认使用反斜杠\作为转义符SELECT U\263A使用自定义转义符SELECT U#263A UESCAPE #。Operations支持的 SQL 特性索引BNF 描述的是批/流查询的特性超集下列专题页对每个特性给出示例并标注该特性仅支持批查询还是流查询或两者皆支持WITH 子句SELECT WHERESELECT DISTINCTWindowing TVF滑动/滚动窗口表值函数窗口聚合分组聚合Over 聚合窗口函数Joins各种连接集合操作UNION / EXCEPT / INTERSECTORDER BY 子句LIMIT 子句Top-NWindow Top-N窗口 Top-NDeduplication去重Pattern Recognition模式匹配Time Travel时间旅行小结围绕这篇官方文档一条可操作的 Flink SQL 开发路径已经完整定义查询用sqlQuery()拿到可组合的Table需要引用外部数据时先注册表/视图或直接用Table.toString()内联背后是UnnamedTable$N临时视图的自动注册执行查询客户端收集结果走executeSql()/Table.execute()得到TableResult落地存储走INSERT INTOTableDescriptor注册的 sink 表结果语义牢记结果只能消费一次与迭代器必须主动关闭两条硬约束并按作业是否开启 exactly-once / at-least-once checkpointing 评估结果的可见时机与重复风险语法边界以 BNF 超集对照所用特性再结合各 Operation 专题页确认该特性在批/流模式下的支持情况避免在流模式下写出只受批支持的语句而触发TableException。赞分享大数据流处理批处理数据工程【免费下载链接】flink项目地址https://gitcode.com/gh_mirrors/fli/flink点击查看免费下载相关推荐Flink SQL 查询概览sqlQuery()/executeSql() 执行、TableResult 结果采集与 SQL 语法全解Flink SQL 查询概览sqlQuery /executeSql 执行、TableResult 结果采集与 SQL 语法全解 本篇技术指南围绕 Apach大数据流处理批处理数据工程Flink SQL DROP 语句完全指南从语法到源码级执行链路解析Flink SQL DROP 语句完全指南从语法到源码级执行链路解析 DROP 语句是 Flink Table SQL API 中用于清理元数据的核心 D大数据流处理批处理数据工程Flink SQL INSERT 语句完全指南从 executeSql 到 StatementSet 的写入实践Flink SQL INSERT 语句完全指南从 executeSql 到 StatementSet 的写入实践 INSERT 语句是 Flink Table大数据流处理批处理数据工程上一篇Park UI 开源项目安装与使用指南下一篇【亲测免费】 whisper-web浏览器中的机器学习驱动语音识别创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考