大數(shù)據(jù)流處理批處理數(shù)據(jù)工程【免費(fèi)下載鏈接】flink項(xiàng)目地址https://gitcode.com/gh_mirrors/fli/flink點(diǎn)擊查看免費(fèi)下載導(dǎo)讀本文是 Apache Flink 中 Table API 與 SQL 的核心入門與深度指南基于倉庫中的官方文檔 docs/content.zh/docs/dev/table/common.md 展開并結(jié)合flink-table模塊的真實(shí)源碼進(jìn)行佐證。Table API 和 SQL 集成在同一套 API 中其核心概念是Table它同時(shí)充當(dāng)查詢的輸入與輸出。讀完本文你將掌握 Table API/SQL 程序的通用結(jié)構(gòu)、如何創(chuàng)建TableEnvironment、如何在 Catalog 中注冊(cè)臨時(shí)表與永久表、如何用 Table API 和 SQL 兩種方式查詢表、如何將結(jié)果輸出到 TableSink以及查詢的翻譯、優(yōu)化與解釋explain機(jī)制從而能夠獨(dú)立搭建從數(shù)據(jù)源到結(jié)果輸出的完整 Flink 表處理程序。Table API 和 SQL 程序的結(jié)構(gòu)所有用于批處理和流處理的 Table API 和 SQL 程序都遵循相同的模式可概括為五個(gè)步驟創(chuàng)建TableEnvironment→ 創(chuàng)建源表source table→ 創(chuàng)建輸出表sink table→ 通過 Table API 或 SQL 構(gòu)建Table查詢 → 將結(jié)果表輸出到 sink。下面的 Java 示例展示了這一通用結(jié)構(gòu)import org.apache.flink.table.api.*; import org.apache.flink.connector.datagen.table.DataGenConnectorOptions; // Create a TableEnvironment for batch or streaming execution. // See the Create a TableEnvironment section for details. TableEnvironment tableEnv TableEnvironment.create(/*…*/); // Create a source table tableEnv.createTemporaryTable(SourceTable, TableDescriptor.forConnector(datagen) .schema(Schema.newBuilder() .column(f0, DataTypes.STRING()) .build()) .option(DataGenConnectorOptions.ROWS_PER_SECOND, 100L) .build()); // Create a sink table (using SQL DDL) tableEnv.executeSql(CREATE TEMPORARY TABLE SinkTable WITH (connector blackhole) LIKE SourceTable (EXCLUDING OPTIONS) ); // Create a Table object from a Table API query Table table1 tableEnv.from(SourceTable); // Create a Table object from a SQL query Table table2 tableEnv.sqlQuery(SELECT * FROM SourceTable); // Emit a Table API result Table to a TableSink, same for SQL result TableResult tableResult table1.insertInto(SinkTable).execute();對(duì)應(yīng)的 Python 版本使用pyflink.table包通過executeSql創(chuàng)建表、from_path讀取表、sql_query執(zhí)行 SQL、execute_insert輸出結(jié)果from pyflink.table import * # Create a TableEnvironment for batch or streaming execution table_env ... # see Create a TableEnvironment section # Create a source table table_env.executeSql(CREATE TEMPORARY TABLE SourceTable ( f0 STRING ) WITH ( connector datagen, rows-per-second 100 ) ) # Create a sink table table_env.executeSql(CREATE TEMPORARY TABLE SinkTable WITH (connector blackhole) LIKE SourceTable (EXCLUDING OPTIONS) ) # Create a Table from a Table API query table1 table_env.from_path(SourceTable).select(...) # Create a Table from a SQL query table2 table_env.sql_query(SELECT ... FROM SourceTable ...) # Emit a Table API result Table to a TableSink, same for SQL result table_result table1.execute_insert(SinkTable)注意示例中datagen連接器用于無界生成測(cè)試數(shù)據(jù)。其rows-per-second選項(xiàng)在源碼中定義于 DataGenConnectorOptions.java默認(rèn)值為10000行/秒見 DataGenConnectorOptionsUtil.java并支持number-of-rows限定總行數(shù)默認(rèn)無限、無界生成。blackhole連接器則是一個(gè)只接收數(shù)據(jù)、不產(chǎn)生任何輸出的“黑洞” sink非常適合快速驗(yàn)證管線正確性。關(guān)于 Scala所有 Flink Scala API 均已棄用deprecated并將在未來的 Flink 版本中移除。你仍然可以用 Scala 構(gòu)建應(yīng)用但建議遷移到 DataStream 和/或 Table API 的 Java 版本。與 DataStream 的集成Table API 和 SQL 查詢可以很容易地集成并嵌入到 DataStream 程序中。請(qǐng)參閱與 DataStream API 集成章節(jié)了解如何將 DataStream 與表之間的相互轉(zhuǎn)化。創(chuàng)建 TableEnvironmentTableEnvironment是 Table API 和 SQL 的核心概念它負(fù)責(zé)在內(nèi)部的 catalog 中注冊(cè)Table注冊(cè)外部的 catalog加載可插拔模塊modules執(zhí)行 SQL 查詢注冊(cè)自定義函數(shù)scalar、table 或 aggregation 函數(shù)DataStream和Table之間的轉(zhuǎn)換面向StreamTableEnvironmentTable總是與特定的TableEnvironment綁定不能在同一條查詢中使用不同TableEnvironment中的表例如對(duì)它們進(jìn)行 join 或 union 操作。TableEnvironment通過靜態(tài)方法TableEnvironment.create()創(chuàng)建該方法在源碼 TableEnvironment.java 中提供兩種重載接收EnvironmentSettings或直接接收Configuration。最常用的方式是基于EnvironmentSettings指定執(zhí)行模式import org.apache.flink.table.api.EnvironmentSettings; import org.apache.flink.table.api.TableEnvironment; EnvironmentSettings settings EnvironmentSettings .newInstance() .inStreamingMode() // 流處理模式 //.inBatchMode() // 批處理模式 .build(); TableEnvironment tEnv TableEnvironment.create(settings);Python 中通過EnvironmentSettings.in_streaming_mode()和EnvironmentSettings.in_batch_mode()分別創(chuàng)建流式與批式環(huán)境from pyflink.table import EnvironmentSettings, TableEnvironment # create a streaming TableEnvironment env_settings EnvironmentSettings.in_streaming_mode() table_env TableEnvironment.create(env_settings) # create a batch TableEnvironment env_settings EnvironmentSettings.in_batch_mode() table_env TableEnvironment.create(env_settings)兩種模式的選擇決定了查詢語義流式模式以無界數(shù)據(jù)流為輸入、支持窗口與狀態(tài)化聚合批式模式則以有界數(shù)據(jù)集為輸入執(zhí)行與傳統(tǒng)數(shù)據(jù)庫類似的批處理語義。從 StreamExecutionEnvironment 創(chuàng)建 StreamTableEnvironment當(dāng)需要與 DataStream API 互操作時(shí)可以從現(xiàn)有的StreamExecutionEnvironment創(chuàng)建StreamTableEnvironmentimport org.apache.flink.streaming.api.environment.StreamExecutionEnvironment; import org.apache.flink.table.api.EnvironmentSettings; import org.apache.flink.table.api.bridge.java.StreamTableEnvironment; StreamExecutionEnvironment env StreamExecutionEnvironment.getExecutionEnvironment(); StreamTableEnvironment tEnv StreamTableEnvironment.create(env);Python 版本同樣支持from pyflink.datastream import StreamExecutionEnvironment from pyflink.table import StreamTableEnvironment s_env StreamExecutionEnvironment.get_execution_environment() t_env StreamTableEnvironment.create(s_env)這樣創(chuàng)建的StreamTableEnvironment既可以通過 Table API/SQL 編寫關(guān)系查詢又可以隨時(shí)將Table與DataStream互相轉(zhuǎn)換實(shí)現(xiàn)關(guān)系查詢與流式算子如 keyBy、map、process的混合編程。在 Catalog 中創(chuàng)建表TableEnvironment維護(hù)著一個(gè)由標(biāo)識(shí)符identifier創(chuàng)建的表 catalog 映射。標(biāo)識(shí)符由三個(gè)部分組成catalog 名稱、數(shù)據(jù)庫名稱以及對(duì)象名稱。如果 catalog 或數(shù)據(jù)庫沒有指明就會(huì)使用當(dāng)前默認(rèn)值參見下文擴(kuò)展表標(biāo)識(shí)符。Table可以是虛擬的視圖VIEWS也可以是常規(guī)的表TABLES視圖VIEWS可以從已經(jīng)存在的Table中創(chuàng)建一般是 Table API 或 SQL 的查詢結(jié)果表TABLES描述的是外部數(shù)據(jù)例如文件、數(shù)據(jù)庫表或消息隊(duì)列。臨時(shí)表Temporary Table和永久表Permanent Table表可以是臨時(shí)的與單個(gè) Flink 會(huì)話session的生命周期相關(guān)也可以是永久的在多個(gè) Flink 會(huì)話和集群cluster中可見。永久表需要 catalog例如 Hive Metastore來維護(hù)表的元數(shù)據(jù)。一旦永久表被創(chuàng)建它將對(duì)任何連接到該 catalog 的 Flink 會(huì)話可見且持續(xù)存在直至被明確刪除。臨時(shí)表通常保存于內(nèi)存中僅在創(chuàng)建它們的 Flink 會(huì)話持續(xù)期間存在對(duì)其它會(huì)話不可見。它們不與任何 catalog 或數(shù)據(jù)庫綁定但可以在一個(gè)命名空間namespace中創(chuàng)建。即使它們對(duì)應(yīng)的數(shù)據(jù)庫被刪除臨時(shí)表也不會(huì)被刪除。屏蔽Shadowing可以使用與已存在的永久表相同的標(biāo)識(shí)符去注冊(cè)臨時(shí)表。此時(shí)臨時(shí)表會(huì)屏蔽永久表并且只要臨時(shí)表存在永久表就無法訪問——所有使用該標(biāo)識(shí)符的查詢都將作用于臨時(shí)表。屏蔽機(jī)制對(duì)實(shí)驗(yàn)experimentation非常有用可以先對(duì)一個(gè)臨時(shí)表執(zhí)行完全相同的查詢例如只包含一個(gè)子集的數(shù)據(jù)或者數(shù)據(jù)是不確定的一旦驗(yàn)證了查詢的正確性就可以對(duì)實(shí)際的生產(chǎn)表進(jìn)行查詢無需修改任何 SQL。創(chuàng)建表虛擬表Virtual Tables在 SQL 的術(shù)語中Table API 的對(duì)象對(duì)應(yīng)于視圖虛擬表。它封裝了一個(gè)邏輯查詢計(jì)劃可以通過以下方式在 catalog 中創(chuàng)建// get a TableEnvironment TableEnvironment tableEnv ...; // see Create a TableEnvironment section // table is the result of a simple projection query Table projTable tableEnv.from(X).select(...); // register the Table projTable as table projectedTable tableEnv.createTemporaryView(projectedTable, projTable);Python 中對(duì)應(yīng)的注冊(cè)方法為register_tableproj_table table_env.from_path(X).select(...) table_env.register_table(projectedTable, proj_table)注意從傳統(tǒng)數(shù)據(jù)庫系統(tǒng)的角度來看Table對(duì)象與VIEW視圖非常像。也就是說定義了Table的查詢沒有被優(yōu)化而且會(huì)被內(nèi)嵌到另一個(gè)引用了這個(gè)注冊(cè)表的查詢中。如果多個(gè)查詢都引用了同一個(gè)注冊(cè)表那么它會(huì)被內(nèi)嵌到每個(gè)查詢中并執(zhí)行多次——注冊(cè)表的結(jié)果不會(huì)被共享。Connector Tables另一種創(chuàng)建TABLE的方式是通過 connector 聲明。Connector 描述了存儲(chǔ)表數(shù)據(jù)的外部系統(tǒng)例如 Apache Kafka 或常規(guī)的文件系統(tǒng)都可以通過這種方式來聲明。這類表既可以通過 Table API 的TableDescriptor直接創(chuàng)建也可以切換到 SQL DDL 創(chuàng)建// Using table descriptors final TableDescriptor sourceDescriptor TableDescriptor.forConnector(datagen) .schema(Schema.newBuilder() .column(f0, DataTypes.STRING()) .build()) .option(DataGenConnectorOptions.ROWS_PER_SECOND, 100L) .build(); tableEnv.createTable(SourceTableA, sourceDescriptor); // 永久表需要 catalog 支持 tableEnv.createTemporaryTable(SourceTableB, sourceDescriptor); // 臨時(shí)表 // Using SQL DDL tableEnv.executeSql(CREATE [TEMPORARY] TABLE MyTable (...) WITH (...));TableDescriptor.forConnector(...)以編程方式構(gòu)建表的連接器類型、schema 與連接選項(xiàng)DataGenConnectorOptions.ROWS_PER_SECOND即源碼中定義的ConfigOption配置鍵rows-per-second使用強(qiáng)類型常量可以避免手寫字符串拼寫錯(cuò)誤。SQL DDL 則提供了與CREATE TABLE一致的聲明式語法二者創(chuàng)建的表可以互換使用。擴(kuò)展表標(biāo)識(shí)符表總是通過三元標(biāo)識(shí)符注冊(cè)包括 catalog 名、數(shù)據(jù)庫名和表名。用戶可以指定一個(gè) catalog 和數(shù)據(jù)庫作為“當(dāng)前 catalog”和“當(dāng)前數(shù)據(jù)庫”這樣三元標(biāo)識(shí)符的前兩個(gè)部分就可以省略未指定時(shí)使用當(dāng)前的 catalog 和當(dāng)前數(shù)據(jù)庫。用戶也可以通過 Table API 或 SQL 切換當(dāng)前的 catalog 和當(dāng)前的數(shù)據(jù)庫。標(biāo)識(shí)符遵循 SQL 標(biāo)準(zhǔn)因此使用時(shí)需要用反引號(hào)進(jìn)行轉(zhuǎn)義。以下 Java 示例演示了不同標(biāo)識(shí)符寫法TableEnvironment tEnv ...; tEnv.useCatalog(custom_catalog); tEnv.useDatabase(custom_database); Table table ...; // register the view named exampleView in the catalog named custom_catalog // in the database named custom_database tableEnv.createTemporaryView(exampleView, table); // register the view named exampleView in the catalog named custom_catalog // in the database named other_database tableEnv.createTemporaryView(other_database.exampleView, table); // register the view named example.View in the catalog named custom_catalog // in the database named custom_database tableEnv.createTemporaryView(example.View, table); // register the view named exampleView in the catalog named other_catalog // in the database named other_database tableEnv.createTemporaryView(other_catalog.other_database.exampleView, table);從源碼接口看useCatalog(String)與useDatabase(String)均定義于 TableEnvironment.java 的TableEnvironment接口中是切換當(dāng)前命名空間的官方 API。Python 中對(duì)應(yīng)為use_catalog(...)與use_database(...)注冊(cè)方法為create_temporary_view(...)。查詢表Table APITable API 是關(guān)于 Java 和 Scala 的集成語言式查詢 API。與 SQL 相反Table API 的查詢不是由字符串指定而是在宿主語言中逐步構(gòu)建。Table API 基于Table類該類表示一個(gè)表流或批處理并提供使用關(guān)系操作的方法。這些方法返回一個(gè)新的Table對(duì)象該對(duì)象表示對(duì)輸入 Table 進(jìn)行關(guān)系操作的結(jié)果。一些關(guān)系操作由多個(gè)方法調(diào)用組成例如table.groupBy(...).select(...)其中g(shù)roupBy(...)指定table的分組而select(...)是在分組上的投影。文檔 Table API 說明了所有流處理和批處理表支持的 Table API 算子。以下示例展示了一個(gè)簡(jiǎn)單的 Table API 聚合查詢——計(jì)算法國所有客戶的收入// get a TableEnvironment TableEnvironment tableEnv ...; // see Create a TableEnvironment section // register Orders table // scan registered Orders table Table orders tableEnv.from(Orders); // compute revenue for all customers from France Table revenue orders .filter($(cCountry).isEqual(FRANCE)) .groupBy($(cID), $(cName)) .select($(cID), $(cName), $(revenue).sum().as(revSum)); // emit or convert Table // execute queryPython 版本使用col(...)引用列orders table_env.from_path(Orders) revenue orders \ .filter(col(cCountry) FRANCE) \ .group_by(col(cID), col(cName)) \ .select(col(cID), col(cName), col(revenue).sum.alias(revSum))注意$(...)是基于字符串的表達(dá)式引用fromDataStream轉(zhuǎn)換或已有表均可使用filter/groupBy/select等算子都返回新的Table因而可以無限鏈?zhǔn)浇M合。SQLFlink SQL 是基于實(shí)現(xiàn)了 SQL 標(biāo)準(zhǔn)的 Apache Calcite 描述了 Flink 對(duì)流處理和批處理表的 SQL 支持。下面的示例演示了如何指定查詢并將結(jié)果作為Table對(duì)象返回// get a TableEnvironment TableEnvironment tableEnv ...; // see Create a TableEnvironment section // register Orders table // compute revenue for all customers from France Table revenue tableEnv.sqlQuery( SELECT cID, cName, SUM(revenue) AS revSum FROM Orders WHERE cCountry FRANCE GROUP BY cID, cName ); // emit or convert Table // execute query如下示例展示了如何指定一個(gè)更新查詢insert query將查詢的結(jié)果插入到已注冊(cè)的表中// get a TableEnvironment TableEnvironment tableEnv ...; // see Create a TableEnvironment section // register Orders table // register RevenueFrance output table // compute revenue for all customers from France and emit to RevenueFrance tableEnv.executeSql( INSERT INTO RevenueFrance SELECT cID, cName, SUM(revenue) AS revSum FROM Orders WHERE cCountry FRANCE GROUP BY cID, cName );Python 對(duì)應(yīng)為sql_query(...)與execute_sql(...)。可以看到sqlQuery用于讀取返回Table對(duì)象而executeSql用于執(zhí)行 DDL/DML如CREATE TABLE、INSERT INTO這是兩者在用法上的關(guān)鍵區(qū)別?；煊?Table API 和 SQLTable API 和 SQL 查詢的混用非常簡(jiǎn)單因?yàn)樗鼈兌挤祷豑able對(duì)象可以在 SQL 查詢返回的Table對(duì)象上定義 Table API 查詢?cè)赥ableEnvironment中注冊(cè)的結(jié)果表可以在 SQL 查詢的FROM子句中引用通過這種方法就可以在 Table API 查詢的結(jié)果上定義 SQL 查詢。輸出表Table通過寫入TableSink輸出。TableSink是一個(gè)通用接口用于支持多種文件格式如 CSV、Apache Parquet、Apache Avro、存儲(chǔ)系統(tǒng)如 JDBC、Apache HBase、Apache Cassandra、Elasticsearch或消息隊(duì)列系統(tǒng)如 Apache Kafka、RabbitMQ。批處理Table只能寫入BatchTableSink流處理Table需要指定寫入AppendStreamTableSink、RetractStreamTableSink或UpsertStreamTableSink之一取決于結(jié)果流的更新模式。請(qǐng)參考文檔 Table Sources Sinks 獲取更多關(guān)于可用 Sink 的信息以及如何自定義DynamicTableSink。方法Table.insertInto(String tableName)定義了一個(gè)完整的端到端管道將源表中的數(shù)據(jù)傳輸?shù)揭粋€(gè)被注冊(cè)的輸出表中。該方法通過名稱在 catalog 中查找輸出表并確認(rèn)Tableschema 與輸出表 schema 一致。從源碼看insertInto返回TablePipeline對(duì)象見 Table.java可以通過TablePipeline.explain()和TablePipeline.execute()分別解釋和執(zhí)行一個(gè)數(shù)據(jù)流管道。下面的示例演示如何輸出Table其中輸出表使用filesystem連接器 CSV 格式并以|作為字段分隔符// get a TableEnvironment TableEnvironment tableEnv ...; // see Create a TableEnvironment section // create an output Table final Schema schema Schema.newBuilder() .column(a, DataTypes.INT()) .column(b, DataTypes.STRING()) .column(c, DataTypes.BIGINT()) .build(); tableEnv.createTemporaryTable(CsvSinkTable, TableDescriptor.forConnector(filesystem) .schema(schema) .option(path, /path/to/file) .format(FormatDescriptor.forFormat(csv) .option(field-delimiter, |) .build()) .build()); // compute a result Table using Table API operators and/or SQL queries Table result ...; // Prepare the insert into pipeline TablePipeline pipeline result.insertInto(CsvSinkTable); // Print explain details pipeline.printExplain(); // emit the result Table to the registered TableSink pipeline.execute();Python 中直接通過result.execute_insert(CsvSinkTable)一步完成插入與執(zhí)行。這里體現(xiàn)了 sink 管道的兩個(gè)階段insertInto只是準(zhǔn)備構(gòu)建TablePipeline可先行printExplain()檢查執(zhí)行計(jì)劃真正觸發(fā)執(zhí)行的是pipeline.execute()。翻譯與執(zhí)行查詢不論輸入數(shù)據(jù)源是流式的還是批式的Table API 和 SQL 查詢都會(huì)被轉(zhuǎn)換成 DataStream 程序。查詢?cè)趦?nèi)部表示為邏輯查詢計(jì)劃并被翻譯成兩個(gè)階段優(yōu)化邏輯執(zhí)行計(jì)劃翻譯成 DataStream 程序Table API 或 SQL 查詢?cè)谙铝星闆r下會(huì)被翻譯TableEnvironment.executeSql()被調(diào)用時(shí)用于執(zhí)行一條 SQL 語句一旦被調(diào)用SQL 語句立即被翻譯TablePipeline.execute()被調(diào)用時(shí)用于執(zhí)行一個(gè)源表到輸出表的數(shù)據(jù)流一旦被調(diào)用Table API 程序立即被翻譯Table.execute()被調(diào)用時(shí)用于將一個(gè)表的內(nèi)容收集到本地一旦被調(diào)用Table API 程序立即被翻譯StatementSet.execute()被調(diào)用時(shí)TablePipeline通過StatementSet.add()輸出給某個(gè) Sink和 INSERT 語句通過調(diào)用StatementSet.addInsertSql()會(huì)先被緩存到StatementSet中當(dāng)StatementSet.execute()被調(diào)用時(shí)所有的 sink 會(huì)被優(yōu)化成一張有向無環(huán)圖DAG從而共享公共子計(jì)劃、避免重復(fù)計(jì)算Table被轉(zhuǎn)換成DataStream時(shí)參閱與 DataStream 集成轉(zhuǎn)換完成后它就成為一個(gè)普通的 DataStream 程序并會(huì)在調(diào)用StreamExecutionEnvironment.execute()時(shí)被執(zhí)行。StatementSet由TableEnvironment.createStatementSet()創(chuàng)建接口定義見 TableEnvironment.java適合在一條作業(yè)中批量提交多個(gè) sink 的場(chǎng)景。查詢優(yōu)化Apache Flink 使用并擴(kuò)展了 Apache Calcite 來執(zhí)行復(fù)雜的查詢優(yōu)化包括一系列基于規(guī)則和基于成本的優(yōu)化例如基于 Apache Calcite 的子查詢解相關(guān)subquery decorrelation投影剪裁projection pruning分區(qū)剪裁partition pruning過濾器下推filter push-down子計(jì)劃消除重復(fù)數(shù)據(jù)以避免重復(fù)計(jì)算特殊子查詢重寫包括兩部分將IN和EXISTS轉(zhuǎn)換為 left semi-joins將NOT IN和NOT EXISTS轉(zhuǎn)換為 left anti-join可選 join 重新排序通過table.optimizer.join-reorder-enabled配置啟用注意當(dāng)前僅在子查詢重寫的結(jié)合條件下支持IN/EXISTS/NOT IN/NOT EXISTS。優(yōu)化器不僅基于計(jì)劃還基于可從數(shù)據(jù)源獲得的豐富統(tǒng)計(jì)信息以及每個(gè)算子例如 io、cpu、網(wǎng)絡(luò)和內(nèi)存的細(xì)粒度成本來做出明智的決策。高級(jí)用戶可以通過CalciteConfig對(duì)象提供自定義優(yōu)化通過調(diào)用TableEnvironment#getConfig#setPlannerConfig將其提供給 TableEnvironment。解釋表ExplainTable API 提供了一種機(jī)制來解釋計(jì)算Table的邏輯和優(yōu)化查詢計(jì)劃。這是通過Table.explain()方法或者StatementSet.explain()方法完成的Table.explain()返回一個(gè)Table的計(jì)劃StatementSet.explain()返回多 sink 計(jì)劃的結(jié)果。它們返回一個(gè)描述三種計(jì)劃的字符串關(guān)系查詢的抽象語法樹the Abstract Syntax Tree即未優(yōu)化的邏輯查詢計(jì)劃優(yōu)化的邏輯查詢計(jì)劃物理執(zhí)行計(jì)劃。此外可以用TableEnvironment.explainSql()方法和TableEnvironment.executeSql()方法支持執(zhí)行一個(gè)EXPLAIN語句獲取邏輯和優(yōu)化查詢計(jì)劃請(qǐng)參閱 EXPLAIN 頁面。以下代碼展示了給定Table使用Table.explain()的示例StreamExecutionEnvironment env StreamExecutionEnvironment.getExecutionEnvironment(); StreamTableEnvironment tEnv StreamTableEnvironment.create(env); DataStreamTuple2Integer, String stream1 env.fromElements(new Tuple2(1, hello)); DataStreamTuple2Integer, String stream2 env.fromElements(new Tuple2(1, hello)); // explain Table API Table table1 tEnv.fromDataStream(stream1, $(count), $(word)); Table table2 tEnv.fromDataStream(stream2, $(count), $(word)); Table table table1 .where($(word).like(F%)) .unionAll(table2); System.out.println(table.explain());上述例子的輸出 Abstract Syntax Tree LogicalUnion(all[true]) :- LogicalFilter(condition[LIKE($1, _UTF-16LEF%)]) : - LogicalTableScan(table[[Unregistered_DataStream_1]]) - LogicalTableScan(table[[Unregistered_DataStream_2]]) Optimized Physical Plan Union(all[true], union[count, word]) :- Calc(select[count, word], where[LIKE(word, _UTF-16LEF%)]) : - DataStreamScan(table[[Unregistered_DataStream_1]], fields[count, word]) - DataStreamScan(table[[Unregistered_DataStream_2]], fields[count, word]) Optimized Execution Plan Union(all[true], union[count, word]) :- Calc(select[count, word], where[LIKE(word, _UTF-16LEF%)]) : - DataStreamScan(table[[Unregistered_DataStream_1]], fields[count, word]) - DataStreamScan(table[[Unregistered_DataStream_2]], fields[count, word])可以看到AST 階段展示的是未經(jīng)優(yōu)化的邏輯算子LogicalUnion、LogicalFilter、LogicalTableScan優(yōu)化后的物理計(jì)劃中LIKE過濾條件已經(jīng)被下推進(jìn)Calc算子執(zhí)行計(jì)劃則給出了最終可執(zhí)行的算子拓?fù)?。explain是排查查詢語義與優(yōu)化效果最直接的工具。多 sink 計(jì)劃的解釋當(dāng)使用StatementSet提交多個(gè) sink 時(shí)可以用StatementSet.explain()觀察多 sink 計(jì)劃——所有 sink 被優(yōu)化成一張有向無環(huán)圖公共子計(jì)劃會(huì)被復(fù)用。以下 Java 示例定義了兩個(gè)文件源和兩個(gè)文件 sinkEnvironmentSettings settings EnvironmentSettings.inStreamingMode(); TableEnvironment tEnv TableEnvironment.create(settings); final Schema schema Schema.newBuilder() .column(count, DataTypes.INT()) .column(word, DataTypes.STRING()) .build(); tEnv.createTemporaryTable(MySource1, TableDescriptor.forConnector(filesystem) .schema(schema) .option(path, /source/path1) .format(csv) .build()); tEnv.createTemporaryTable(MySource2, TableDescriptor.forConnector(filesystem) .schema(schema) .option(path, /source/path2) .format(csv) .build()); tEnv.createTemporaryTable(MySink1, TableDescriptor.forConnector(filesystem) .schema(schema) .option(path, /sink/path1) .format(csv) .build()); tEnv.createTemporaryTable(MySink2, TableDescriptor.forConnector(filesystem) .schema(schema) .option(path, /sink/path2) .format(csv) .build()); StatementSet stmtSet tEnv.createStatementSet(); Table table1 tEnv.from(MySource1).where($(word).like(F%)); stmtSet.add(table1.insertInto(MySink1)); Table table2 table1.unionAll(tEnv.from(MySource2)); stmtSet.add(table2.insertInto(MySink2)); String explanation stmtSet.explain(); System.out.println(explanation);Python 中對(duì)應(yīng)為create_statement_set()、stmt_set.add_insert(MySink1, table1)與stmt_set.explain()。多 sink 計(jì)劃的輸出節(jié)選 Abstract Syntax Tree LogicalLegacySink(name[default_catalog.default_database.MySink1], fields[count, word]) - LogicalFilter(condition[LIKE($1, _UTF-16LEF%)]) - LogicalTableScan(table[[default_catalog, default_database, MySource1, source: [CsvTableSource(read fields: count, word)]]]) LogicalLegacySink(name[default_catalog.default_database.MySink2], fields[count, word]) - LogicalUnion(all[true]) :- LogicalFilter(condition[LIKE($1, _UTF-16LEF%)]) : - LogicalTableScan(table[[default_catalog, default_database, MySource1, source: [CsvTableSource(read fields: count, word)]]]) - LogicalTableScan(table[[default_catalog, default_database, MySource2, source: [CsvTableSource(read fields: count, word)]]]) Optimized Physical Plan LegacySink(name[default_catalog.default_database.MySink1], fields[count, word]) - Calc(select[count, word], where[LIKE(word, _UTF-16LEF%)]) - LegacyTableSourceScan(table[[default_catalog, default_database, MySource1, source: [CsvTableSource(read fields: count, word)]]], fields[count, word]) LegacySink(name[default_catalog.default_database.MySink2], fields[count, word]) - Union(all[true], union[count, word]) :- Calc(select[count, word], where[LIKE(word, _UTF-16LEF%)]) : - LegacyTableSourceScan(table[[default_catalog, default_database, MySource1, source: [CsvTableSource(read fields: count, word)]]], fields[count, word]) - LegacyTableSourceScan(table[[default_catalog, default_database, MySource2, source: [CsvTableSource(read fields: count, word)]]], fields[count, word]) Optimized Execution Plan Calc(select[count, word], where[LIKE(word, _UTF-16LEF%)])(reuse_id[1]) - LegacyTableSourceScan(table[[default_catalog, default_database, MySource1, source: [CsvTableSource(read fields: count, word)]]], fields[count, word]) LegacySink(name[default_catalog.default_database.MySink1], fields[count, word]) - Reused(reference_id[1]) LegacySink(name[default_catalog.default_database.MySink2], fields[count, word]) - Union(all[true], union[count, word]) :- Reused(reference_id[1]) - LegacyTableSourceScan(table[[default_catalog, default_database, MySource2, source: [CsvTableSource(read fields: count, word)]]], fields[count, word])注意執(zhí)行計(jì)劃中的reuse_id[1]與Reused(reference_id[1])由于MySink1與MySink2都引用了對(duì)MySource1的相同過濾查詢StatementSet優(yōu)化器將其識(shí)別為公共子計(jì)劃并復(fù)用避免了同一份Calc計(jì)算被重復(fù)執(zhí)行——這正是StatementSet相比多次單獨(dú)executeSql的核心優(yōu)勢(shì)之一?？偨Y(jié)圍繞Table這一核心概念Flink Table API 與 SQL 形成了完整一致的編程模型通過TableEnvironment統(tǒng)一管理 catalog、模塊與函數(shù)注冊(cè)通過臨時(shí)/永久表、虛擬視圖與 connector 表三種形態(tài)描述數(shù)據(jù)以 Table API 或 SQL 兩種等價(jià)方式構(gòu)建查詢經(jīng)insertInto/TablePipeline將結(jié)果輸出到 sink最終由基于 Calcite 的優(yōu)化器完成兩階段翻譯與優(yōu)化并在Table.execute()、TablePipeline.execute()、StatementSet.execute()等觸發(fā)點(diǎn)真正執(zhí)行。explain機(jī)制則為調(diào)試與優(yōu)化提供了從 AST 到物理執(zhí)行計(jì)劃的全程可視化。本文所涉及的源碼證據(jù)均可在倉庫flink-table與flink-connectors模塊中找到讀者可結(jié)合 docs/content.zh/docs/dev/table/tableApi.md、docs/content.zh/docs/dev/table/sql/overview.md 與 docs/content.zh/docs/dev/table/catalogs.md 繼續(xù)深入。贊分享大數(shù)據(jù)流處理批處理數(shù)據(jù)工程【免費(fèi)下載鏈接】flink項(xiàng)目地址https://gitcode.com/gh_mirrors/fli/flink點(diǎn)擊查看免費(fèi)下載相關(guān)推薦終極百度貼吧個(gè)性化體驗(yàn)指南TiebaTS模塊完全解析終極百度貼吧個(gè)性化體驗(yàn)指南TiebaTS模塊完全解析 TiebaTS是一款基于Xposed框架的開源百度貼吧增強(qiáng)模塊專為追求純凈、高效貼吧瀏覽體驗(yàn)的用戶設(shè)計(jì)后端運(yùn)維觀測(cè)告警可觀測(cè)性人工智能AI Agent3個(gè)核心策略深度優(yōu)化MediaPipe GPU性能的完整指南3個(gè)核心策略深度優(yōu)化MediaPipe GPU性能的完整指南 MediaPipe作為跨平臺(tái)的機(jī)器學(xué)習(xí)框架為實(shí)時(shí)媒體處理提供了強(qiáng)大的GPU加速能力。在追求極致人工智能機(jī)器學(xué)習(xí)計(jì)算機(jī)視覺多模態(tài)本地部署Flink SQL JSON完全指南從解析到嵌套查詢實(shí)戰(zhàn)Flink SQL JSON完全指南從解析到嵌套查詢實(shí)戰(zhàn) 你是否還在為JSON數(shù)據(jù)解析頭疼面對(duì)多層嵌套結(jié)構(gòu)無從下手本文將帶你掌握Apache Flink大數(shù)據(jù)流處理批處理數(shù)據(jù)工程上一篇Vue Router 導(dǎo)航故障Navigation Failures完全指南檢測(cè)、分類與實(shí)戰(zhàn)處理下一篇從惱人廣告到絲滑觀看Improve YouTube! 擴(kuò)展的六種打開方式創(chuàng)作聲明：本文部分內(nèi)容由AI輔助生成（AIGC），僅供參考