Flink SQL- FineDatalink 幫助文檔

文檔創建者：帆软用户UT9OHLQhgP

編輯次數：1

最近更新：帆软用户UT9OHLQhgP 于 2026-09-02

提示:

產品註冊版本為「v5.0」時，才能使用資料開發-實時任務。

1. 概述

1.1 版本

FineDataLink 版本	功能變動
5.0.1.4	支援使用 Flink SQL 算子
5.0.1.5	實時任務中若使用 WebSocket輸入、MQTT輸入、IBM MQ輸入、Pulsar輸入，不能再使用「Flink SQL」算子
5.0.11.2	新增「事件時間」欄位，可以定義事件時間，滿足使用者基於事件時間的轉換操作
5.0.12.3	「Flink SQL」算子前，可接入多個上游算子

1.2 應用場景

實時任務發展前期，視覺化算子的單行轉換能力、多行聚合能力不足以滿足客戶各式各樣的轉換需求。
使用者習慣使用 SQL 語句進行資料開發，但目前實時任務中不支援編寫 SQL 語句處理資料。

1.3 功能簡介

新增「Flink SQL」算子，使用者可編寫 SQL 語句在實時任務中對資料進行查詢和處理。如下圖所示：

1.4 前提條件

使用「Flink SQL」算子前，需要：

註冊「實時進階算子包」功能點，詳情請參見：註冊簡介
需要配置 Flink引擎

2. 功能說明

2.1 通用說明

1）「Flink SQL」算子前必須有前置算子：

FineDataLink 版本	說明
5.0.12.3 之前版本	「Flink SQL」算子前只能接一個輸入算子
5.0.12.3 及之後版本	「Flink SQL」算子前可接多個輸入算子

2）「Flink SQL」算子後可接多個匯出算子。

3）5.0.1.5 及之後版本，實時任務中若使用 WebSocket輸入、MQTT輸入、IBM MQ輸入、Pulsar輸入，不能再使用「Flink SQL」節點。

2.2 SQL語句

注：不支援在一個「Flink SQL」算子中，使用多個 select 語句。

設定項	說明
函式	支援的函式：Flink相關函式
支援、不支援語句	支援語句：select、with、where、distinct、tumble、table、hop、 cumulate、 session、 grouping sets、rollup、cube、 group by、 having、over、order by
支援、不支援語句	不支援語句：semi、union、intersect、except、in 、exists、limit、create、drop、alter、Insert、analyze、describe、truncate、explain、use、show、load、show、load、unload、set、reset、jar、job、update、delete、call
使用方式	1）在 SQL 中，可以透過 ${valueName} 的方式，使用自訂的任務參數 2）在輸入 SQL 語句時有聯想功能，支援聯想關鍵詞、函式、任務靜態參數、上游算子

2.3 事件時間欄位

5.0.11.2 及之後版本新增該欄位。

2.3.1 應用場景

使用者需要基於 FDL，進行銷售分析，統計每 10 分鐘內每種商品售賣的情況，資料來自於 CRM 系統資料庫。
需要基於裝置的原始資料進行裝置監視，資料來源是 Kafka，想要分析每 1 分鐘裝置的平均溫度。

5.0.11.2 之前版本，使用者基於時間視窗的匯總計算時，由於沒有定義事件時間，無法進行轉換，Flink SQL 會進行報錯。

2.3.2 功能說明

事件時間欄位說明：

設定項	說明
欄位名	預設為event_time，支援修改
欄位類型	預設為timestamp(3)，不支援使用者修改當「來自上游欄位」類型為timestamp（3），event_time 欄位類型預設邏輯：複製原始欄位當「來自上游欄位」類型不為timestamp(3)，event_time欄位類型預設邏輯：複製原始欄位並轉換類型為 TIMESTAMP(3) 當「來自上游欄位」精度不為 3 時，event_time欄位預設邏輯：精度超過3，捨棄為3，精度小於3，補齊為3
來自上游欄位	1）僅允許選擇一個欄位作為事件時間欄位；欄位選擇為實際的業務欄位，即資料中含有的欄位，是業務發生的時間，而不是資料產生的時間 2）支援的欄位類型：數值類型：bigint、long 字元類型：string、varchar 日期類型：date、time、timestamp、timestamp_ltz 3）選擇上游欄位後，工程會直接定義watermark，預設值為 30 s；使用者可修改 FineDB 參數來修改 watermark，具體操作請聯絡技術支援

設定項

說明

欄位名

預設為event_time，支援修改

欄位類型

預設為timestamp(3)，不支援使用者修改

當「來自上游欄位」類型為timestamp（3），event_time 欄位類型預設邏輯：複製原始欄位
當「來自上游欄位」類型不為timestamp(3)，event_time欄位類型預設邏輯：複製原始欄位並轉換類型為 TIMESTAMP(3)
當「來自上游欄位」精度不為 3 時，event_time欄位預設邏輯：精度超過3，捨棄為3，精度小於3，補齊為3

來自上游欄位

1）僅允許選擇一個欄位作為事件時間欄位；欄位選擇為實際的業務欄位，即資料中含有的欄位，是業務發生的時間，而不是資料產生的時間

2）支援的欄位類型：

數值類型：bigint、long
字元類型：string、varchar
日期類型：date、time、timestamp、timestamp_ltz

3）選擇上游欄位後，工程會直接定義watermark，預設值為 30 s；使用者可修改 FineDB 參數來修改 watermark，具體操作請聯絡技術支援

SQL語句說明：

需要先定義「事件時間欄位」，再在 SQL 語句中使用視窗函式：

函式類別	函式全名/文法	功能描述	SQL範例
視窗定義	捲動視窗/tumble	TUMBLE 函式指定每個元素到一個指定大小的視窗中。捲動視窗的大小固定且不重複。例如：假設指定了一個 5 分鐘的捲動視窗。Flink 將每 5 分鐘生成一個新的視窗	SELECT * FROM TUMBLE(TABLE Bid, DESCRIPTOR(bidtime), INTERVAL '10' MINUTES);
	滑動視窗/hop	滑動視窗函式指定元素到一個定長的視窗中。和捲動視窗很像，有視窗大小參數，另外增加了一個視窗滑動步長參數。如果滑動步長小於視窗大小，就能產生資料重疊的效果。在這個例子裏，資料可以被分配在多個視窗	SELECT * FROM HOP(TABLE Bid, DESCRIPTOR(bidtime), INTERVAL '5' MINUTES, INTERVAL '10' MINUTES);
	累計視窗/cumulate	CUMULATE 函式指定元素到多個視窗，從初始的視窗開始，直到達到最大的視窗大小的視窗，所有的視窗都包含其區間內的元素，另外，視窗的開始時間是固定的。你可以將 CUMULATE 函式視為首先應用具有最大視窗大小的 TUMBLE 視窗，然後將每個捲動視窗拆分為具有相同視窗開始但視窗結束步長不同的幾個視窗。所以累積視窗會產生重疊並且沒有固定大小	SELECT * FROM CUMULATE(TABLE Bid, DESCRIPTOR(bidtime), INTERVAL '2' MINUTES, INTERVAL '10' MINUTES);
視窗聚合		視窗聚合是透過 GROUP BY 子句定義的，其特徵是包含視窗表值函式產生的 “window_start” 和 “window_end” 列。和普通的 GROUP BY 子句一樣，視窗聚合對於每個組會計算出一行資料	捲動視窗：SELECT window_start, window_end, SUM(price) AS total_price FROM TUMBLE(TABLE Bid, DESCRIPTOR(bidtime), INTERVAL '10' MINUTES) GROUP BY window_start, window_end; 滑動視窗：SELECT window_start, window_end, SUM(price) AS total_price FROM HOP(TABLE Bid, DESCRIPTOR(bidtime), INTERVAL '5' MINUTES, INTERVAL '10' MINUTES) GROUP BY window_start, window_end; 累計視窗：SELECT window_start, window_end, SUM(price) AS total_price FROM CUMULATE(TABLE Bid, DESCRIPTOR(bidtime), INTERVAL '2' MINUTES, INTERVAL '10' MINUTES) GROUP BY window_start, window_end;
視窗 TOP-N		視窗 Top-N 是特殊的 Top-N，它傳回每個分割槽鍵的每個視窗的N個最小或最大值。	SELECT [column_list] FROM ( SELECT [column_list], ROW_NUMBER() OVER (PARTITION BY window_start, window_end [, col_key1...] ORDER BY col1 [asc\|desc][, col2 [asc\|desc]...]) AS rownum FROM table_name) -- relation applied windowing TVF WHERE rownum <= N [AND conditions]
視窗去重		視窗去重是一種特殊的去重，它根據指定的多個列來刪除重複的行，保留每個視窗和分割槽鍵的第一個或最後一個資料	SELECT * FROM ( SELECT bidtime, price, item, supplier_id, window_start, window_end, ROW_NUMBER() OVER (PARTITION BY window_start, window_end ORDER BY bidtime DESC) AS rownum FROM TABLE( TUMBLE(TABLE Bid, DESCRIPTOR(bidtime), INTERVAL '10' MINUTES)) ) WHERE rownum <= 1;

3. 操作步驟

本文範例：使用「CDC輸入」算子取出 MySQL 資料，使用「Flink SQL」算子將取出的 name 欄位根據不同值進行分組，並統計每個不同 name 欄位出現的次數，傳回 name 值和對應的出現次數。

3.1 設定前置算子

1）建立實時任務。

2）拖入一個「CDC輸入」算子，取出待處理資料。如下圖所示：

點選「資料預覽」，可查看取出的資料。如下圖所示：

3.2 設定 Flink SQL 算子

拖入「Flink SQL」算子，將取出的 name 欄位根據不同值進行分組，並統計每個不同 name 欄位出現的次數，傳回 name 值和對應的出現次數。如下圖所示：

點選「資料預覽」，如下圖所示：

3.3 後續步驟

使用者可繼續處理資料，或者後接匯出算子將資料匯出。

注：「Flink SQL」算子後可接多個匯出算子。

已經是第一篇

已經是最後一篇

有幫助
沒幫助
只是瀏覽

中文（简体） English

中文（繁體）

Flink SQL