用SQL分析數據流

越來越多的公司在采用流處理技術，並將現有的批處理應用程序遷移到流處理或者為新的應用設計流處理方案。其中許多應用程序專註於分析流數據。分析的數據流來源廣泛，如數據庫交易，點擊，傳感器測量或物聯網設備。

Apache Flink非常適合流式分析，因為它提供了事件時間語義支持，恰一次的處理，並同時實現了高吞吐和低延遲。由於這些特性，Flink能夠近乎實時地從大量輸入流計算確切的和確定性的結果，同時在出現故障時提供恰一次處理的語義。

Flink的流處理核心API，DataStream API，非常具有表現力，並為許多常見操作提供原語。除了其他功能之外，它還提供高度可定制的窗口邏輯，具有不同性能特性的不同狀態原語，用於註冊和響應定時器的鉤子，以及用於向外部系統提供高效異步請求的工具。另一方面，許多流分析應用程序遵循類似的模式，並且不需要DataStream API提供的表達級別。他們可以使用特定領域語言以更自然和簡潔的方式表達。眾所周知，SQL是數據分析的事實標準。對於流式分析，SQL可以讓更多的人在更短的時間內在數據流上開發應用程序。但是，還沒有開源流處理器提供全面良好的SQL支持。

為什麽Streams上的SQL是一個大問題？

由於許多原因，SQL是數據分析中使用最廣泛的語言：

因此，能夠使用SQL處理和分析數據流，使流處理技術可供更多用戶使用。此外，由於SQL的聲明性和自動優化的潛力，它大大減少了開發高效流分析應用程序的時間和精力。

但是，SQL（以及關系數據模型和代數）設計的時候並沒有考慮到流式數據。關系是（多）集合，而不是無限的元組序列。在執行SQL查詢時，傳統的數據庫系統和查詢引擎將讀取並處理完整可用的數據集，並生成固定大小的結果。相反，數據流不斷提供新的記錄，使得數據隨著時間的推移而到達。因此，流式查詢必須持續處理到達的數據，而不是“完整的數據”。

這就是說，用SQL處理流並不是不可能的。一些關系數據庫系統具有物化視圖的急切維護功能，這類似於評估數據流上的SQL查詢。物化視圖與常規（虛擬）視圖一樣被定義為SQL查詢。但是，物化視圖查詢的結果實際上是存儲（或物化）在內存或磁盤上的，這樣查詢不需要在查詢時即時計算。為了防止物化視圖變舊，數據庫系統需要在其基本關系（定義查詢中引用的表）被修改時更新視圖。如果將視圖基礎關系的修改視為修改流(或者視為變更日誌流)，很明顯就是在流上的物化視圖為何和sql在某種程度上是相關的。

FlinkAPI：表API和SQL

自2016年8月發布1.1.0版本以來，Flink具有兩種語義等效的關系API，嵌入語言（language-embedded）的Table API（用於Java和Scala）和標準SQL。這兩個API都被設計為實時處理和離線批處理的統一API。這意味著，

無論其輸入是靜態批量數據還是流式數據，查詢都會產生完全相同的結果。

出於多種原因，流和批處理的統一API非常重要。首先，用戶只需要學習一個API來處理靜態和流式數據。此外，可以使用相同的查詢來分析批量和流式數據，從而可以在同一查詢中共同時分析歷史數據和實時數據。在目前的狀態下，我們尚未實現批量和流式語義的完全統一，但社區在實現這一目標方面正取得很好的進展。

以下代碼片段顯示了兩個等效的Table API和SQL查詢，這些查詢計算溫度傳感器測量流上簡單的窗口集合。SQL查詢的語法基於Apache Calcite的分組窗口函數的語法，並將在Flink的1.3.0版中得到支持。

正如您所看到的，這兩個API都彼此緊密集成，並與Flink的主要DataStream和DataSet API 緊密集成。一個Table可以生成於一個DataSet或DataStream，也可以轉換成一個DataSet或DataStream。因此，可以輕松掃描外部表源（如數據庫或Parquet文件），使用Table API查詢執行一些預處理，將結果轉換為DataSet並在其上運行Gelly圖算法。以上示例中定義的查詢也可以用於通過更改執行環境來處理批處理數據。

在內部，兩個API都被翻譯成相同的邏輯表示，並由Apache Calcite進行優化，並編譯到DataStream或DataSet程序中。實際上，優化和編譯過程並不知道查詢是使用Table API還是SQL來定義的。由於Table API和SQL在語義方面是等價的，而且只有語法不同，所以當我們在這篇文章中討論SQL時，我們總是引用這兩個API。

在當前狀態（版本1.2.0）中，Flink的關系API支持數據流上有限的一組關系運算符，包括projections，過濾器和窗口聚合（projections, filters, and windowed aggregates）。所有支持的操算子都有共同之處：他們從不更新已經發布的結果記錄。對於projection and filter等一次性記錄操作算子來說，這顯然不是問題。但是，它會影響收集和處理多個記錄的操作算子，例如窗口聚合。由於發布的結果無法更新，因此在Flink 1.2.0中必須丟棄在結果發布後到達的輸入記錄。

對於向存儲系統發送數據的應用程序（如Kafka主題，消息隊列或僅支持追加操作且不更新或刪除的文件），當前版本的限制是可接受的。遵循此模式的常見用例是例如連續ETL和流歸檔應用程序，這些應用程序將流保存到歸檔或為進一步聯機（流式）分析或後續離線分析準備數據。由於無法更新之前發布的結果，因此這些類型的應用程序必須確保發布的結果是正確的，並且將來不需要進行更正。下圖說明了這些應用程序。

雖然僅支持追加的查詢對於某些類型的應用程序和特定類型的存儲系統很有用，但有很多流分析用例需要更新結果。這包括流式處理應用程序，這些應用程序不能丟棄遲到的記錄，需要（長時間運行的）窗口化聚合的早期結果，或需要非窗口聚合。在每種情況下，以前發出的結果記錄都需要更新。結果更新查詢通常會將其結果實現為外部數據庫或鍵值存儲，以便外部應用程序可以訪問並進行查詢。實現這種模式的應用程序是儀表板，報告應用程序或其他應用程序，這需要及時獲得不斷更新的結果。下圖說明了這些類型的應用程序。

動態表上的連續查詢

支持更新先前發布結果的查詢是Flink關系API的下一個重要步驟。此功能非常重要，因為它大大增加了API的範圍和支持的用例範圍。

因此，當添加對結果更新查詢的支持時，我們當然必須保留流和批輸入的統一語義。我們通過動態表的概念來實現這一點。動態表是一個不斷更新的表，可以像常規的靜態表一樣查詢。但是，與作為結果終止並返回靜態表的批處理表相比，對動態表的查詢連續運行，並生成一個根據輸入表上的修改不斷更新的表。因此，結果表也是一個動態表。這個概念與我們之前討論的物化視圖維護非常相似。

假設我們可以在產生新動態表的動態表上運行查詢，下一個問題是，流和動態表如何相互關聯？答案是可以將流轉換為動態表，並將動態表轉換為流。下圖顯示了在流上處理關系查詢的概念模型。

首先，將流轉換為動態表。使用連續查詢來查詢動態表，從而生成新的動態表。最後，結果表轉換回流。需要註意的是，這只是邏輯模型，並不意味著查詢是如何實際執行的。實際上，連續查詢在內部翻譯成傳統的DataStream程序。

在下面，我們描述這個模型的不同步驟：

在流上定義動態表

評估動態表上的SQL查詢的第一步是在流上定義一個動態表。這意味著我們必須指定流的記錄如何修改動態表。流攜帶的記錄必須有一個schema，該schema可以映射到表的關系schema。有兩種模式可以在流上定義動態表：追加模式和更新模式。

在追加模式下，每個流記錄都是對動態表的插入修改。因此，流的所有記錄都會追加到動態表中，使其不斷增長並且大小無限。下圖說明了追加模式。

在更新模式下，流記錄可以表示對動態表的插入，更新或刪除修改（追加模式實際上是更新模式的特例）。當通過更新模式在流上定義動態表時，我們可以在表上指定唯一的鍵屬性。在這種情況下，更新和刪除操作是針對key屬性執行的。更新模式在下圖中顯示。

查詢動態表

一旦我們定義了一個動態表，我們就可以在其上運行查詢。由於動態表隨時間而改變，因此我們必須定義查詢動態表的含義。讓我們想象一下，我們在特定的時間點拍攝動態表格的快照。此快照可以視為常規靜態批處理表。我們將動態表A在點t處的快照表示為A [t]。快照可以用任何SQL查詢來查詢。查詢生成一個常規的靜態表作為結果。我們將在時間t的動態表A上的查詢q的結果表示為q（A [t]）。如果我們重復計算查詢動態表快照的結果以獲得進展時間點，我們將獲得許多隨時間變化的靜態結果表，並有效地構成一個動態表。我們在動態表中定義一個查詢的語義如下。

動態表A上的查詢q產生動態表R，其在每個時間點t等於在A [t]上應用q的結果，即R [t]=q（A [t]）。這一定義意味著在一個批處理表上運行在相同的查詢q，並在流表產生相同的結果。在下面，我們給出兩個例子來說明動態表上查詢的語義。

在下圖中，我們在左側看到一個動態輸入表A，它在追加模式下定義。在t=8時，A由六行（藍色）組成。在時間t=9和t=12，分別有一行被追加到A（分別以綠色和橙色顯示）。我們在表A上運行一個圖中心顯示的簡單的查詢。查詢按屬性k分組並統計每組的記錄。在右側，我們看到在時間t=8（藍色），t=9（綠色）和t=12時查詢q的結果（橙子）。在時間t的每個時間點，結果表等同於在時間t時動態表A上的批量查詢。

這個例子中的查詢是一個簡單的分組（但沒有窗口）聚合查詢。因此，結果表的大小取決於輸入表的不同分組鍵的數量。此外，值得註意的是，查詢不斷更新它先前發出的結果行，而不是僅添加新行。

第二個例子展示了一個類似的查詢，它在一個重要方面有所不同 除了在關鍵屬性k上進行分組之外，查詢還將記錄分組到五秒鐘的滾動窗口中，這意味著它計算每五秒每個k值的計數。再次，我們使用Calcite的組窗口函數來指定此查詢。在圖的左側，我們看到輸入表A以及它在追加模式下隨時間變化的情況。在右側，我們看到結果表以及它隨著時間的變化。

與第一個例子的結果相反，結果表相對於時間增長，即每5秒鐘計算一次新的結果行（假設輸入表在過去5秒內接收到更多記錄）。盡管非窗口化查詢（主要）更新結果表的行，但窗口化聚合查詢僅將新行追加到結果表中。

盡管這篇博文主要關註動態表上的SQL查詢的語義，而不是關於如何有效地處理這樣的查詢，但我們想指出，每當更新輸入表時，不可能從頭開始計算查詢的完整結果。相反，查詢被編譯為一個流式處理程序，它根據輸入的變化不斷更新其結果。這意味著並非所有有效的SQL查詢都受支持，但只有那些可以連續，增量和有效計算的SQL查詢才受支持。我們計劃在後續博客文章中討論有關動態表上SQL查詢評估的詳細信息。

發出動態表格

查詢動態表將生成另一個動態表，它表示查詢的結果。根據查詢及其輸入表，結果表通過插入，更新和刪除來持續修改，就像常規數據庫表一樣。它可能是一個帶有單個行的表，它不斷更新，只有插入表而沒有更新修改，或者兩者都有。

傳統數據庫系統在發生故障和復制時使用日誌來重建表。有不同的日誌記錄技術，如UNDO，REDO和UNDO / REDO日誌記錄。簡而言之，UNDO日誌記錄修改元素的先前值以恢復未完成的事務，REDO日誌記錄已修改元素的新值以redo丟失的已完成事務的更改，UNDO / REDO日誌記錄一個變更的元素舊值和新值來撤消未完成的事務和redo已完成的事務的丟失變更。根據這些日誌記錄技術的原理，可以將動態表格轉換為兩種類型的更新日誌流，即REDO流和REDO + UNDO流。

通過將表中的修改轉換為流消息，將動態表轉換為redo+undo流。插入被發射作為帶新的行的插入消息，刪除修改被發射作為帶有舊的行的刪除消息，並且更新修改被發射作為帶有舊的行的刪除消息，並且與新的行的插入消息。下圖說明了此行為。

左邊顯示了一個動態表格，該表格以追加模式維護，並作為圖中心查詢的輸入。查詢結果轉換為底部顯示的redo + undo流。輸入表的第一條記錄（1，A）會在結果表中產生一條新記錄，並因此在流中插入消息+（A，1）。具有k=‘A‘ （4，A）的第二輸入記錄在結果表中產生（A，1）記錄的更新，並因此產生刪除消息- （A，1）和插入消息+（A ，2）。所有下遊操作算子或數據接收器都需要能夠正確處理這兩種類型的消息。

在兩種情況下，動態表可以轉換為redo流：它可以是僅追加表（即僅具有插入修改），也可以具有唯一鍵屬性。動態表上的每個插入修改都會生成一條插入消息，並將新行添加到redo流中。由於redo流的限制，只有具有唯一鍵的表可以進行更新和刪除修改。如果從鍵控動態表中刪除鍵，或者因為行被刪除或因為行的鍵屬性被修改了，則刪除鍵中的刪除鍵被發送到redo流。更新修改產生帶有更新的更新消息，即新行。由於刪除和更新修改是針對唯一key定義的，因此下遊操作員需要能夠通過key訪問先前的值。下圖，展示了相同查詢的結果表是如何轉化為一個redo流的。

產生插入到動態表中的行（1，A）導致+（A，1）插入消息。產生更新的行（4，A）產生*（A，2）更新消息。

redo流的常見用例是將查詢結果寫入僅追加存儲系統，如滾動文件或Kafka主題，或者寫入具有key訪問特性的數據存儲區，如Cassandra，關系型數據庫或壓縮kafka話題。還可以將動態表實現為流式應用程序內部的keyed狀態，以評估連續查詢並使其可從外部系統進行查詢。通過這種設計，Flink自身維護流中持續SQL查詢的結果，並在結果表上提供key查找，例如從儀表板應用程序中進行查找。

切換到動態表格後會發生什麽變化？

在版本1.2中，Flink的關系API的所有流式運算符（如過濾器，項目和組窗口聚合）僅發出新行並且無法更新以前發出的結果。相比之下，動態表格能夠處理更新和刪除修改。現在你可能會問自己：當前版本的處理模型與新的動態表模型有什麽關系？API的語義是否會徹底改變？我們是否需要從頭開始重新實現API以實現所需的語義？

所有這些問題的答案都很簡單。當前的處理模型是動態表模型的一個子集。使用我們在這篇文章中介紹的術語，當前模型將流轉換為追加模式下的動態表格，即無限增長的表格。由於所有運算符只接受插入更改並在其結果表上產生插入更改（即發出新行），所有受支持的查詢都會生成動態追加表，這些追加表將使用redo模型轉換回DataStreams，用於追加表。因此，當前模型的語義被新的動態表模型完全覆蓋和保存。

結論和展望

Flink的關系型API能夠很快實施流分析應用程序並用於多種生產環境。在這篇博文中，我們討論了Table API和SQL的未來。這一努力將使更多人能夠訪問Flink和流處理。此外，用於查詢歷史和實時數據的統一語義以及查詢和維護動態表的概念將使許多令人興奮的用例和應用程序的實現變得非常容易。由於本文主要關註流和動態表上的關系查詢的語義，因此我們沒有討論如何執行查詢的詳細信息，其中包括內部執行回收，處理遲發事件，支持早期結果以及邊界空間要求。

最近幾個月，Flink社區的許多成員一直在討論和貢獻關系API。迄今為止我們取得了很大的進展 雖然大多數工作都側重於以追加模式處理流，但議程上的下一步是處理動態表以支持更新其結果的查詢。倒裝句如果您對使用SQL處理流的想法感到興奮並希望為此付出努力，請提供反饋，加入郵件列表中的討論，或者抓住JIRA問題進行工作。

原文閱讀，請點擊閱讀原文。

推薦閱讀：

1，Spark Streaming 中管理 Kafka Offsets 的幾種方式

2，Flink DataSet編程指南-demo演示及註意事項

3，構建Flink工程及demo演示

4，Flink系列之時間

如果，Google 早已解決不了你的問題。

如果，你還想知道 Apple、Facebook、IBM、阿裏等國內外名企的核心架構設計。

來，我們在深圳準備了知識星球，想助你成長：

文章來源：https://blog.csdn.net/rlnLo2pNEfx9c/article/details/80796601

Flink：動態表上的連續查詢