基于Flink流處理框架的FFT并行及優(yōu)化*

鐘旭陽 ，徐 云

（1.中國科學技術大學 計算機科學與技術學院，安徽 合肥230026；2.安徽省高性能計算重點實驗室，安徽 合肥230026）

0 引言

快速傅里葉變換（Fast Fourier Transform，F(xiàn)FT）是實現(xiàn)離散傅里葉變換及其逆變換的算法。FFT使用分而治之的主要思想，其主要目的是將一個復雜的大問題分解成多個簡單的小問題，然后分別解決這些小問題[1]。FFT在科學計算領域具有極其重要的地位[2]。利用FFT能夠在計算離散傅里葉變換時大大減少所需要的乘法次數(shù)，并且FFT點數(shù)規(guī)模越大，F(xiàn)FT算法所能夠節(jié)省的計算量就越顯著，因此FFT廣泛應用于數(shù)據(jù)信號處理、地震預報、石油勘探等領域。

已有的FFT分布式計算方法大多基于MapReduce批處理系統(tǒng)[1，3-5]，其中 FFT計算作為一個整體，在某一個轉換操作中直接計算來自上一個操作的整個輸出數(shù)據(jù)，忽視了FFT計算特性的同時，還需要等待較長時間才能延遲得到處理結果。目前并未有成熟的、基于流粒度的對FFT的流處理分布式算法并行優(yōu)化相關研究。且現(xiàn)如今Flink分布式流處理框架大都用于社交網絡等領域中簡單的數(shù)據(jù)項統(tǒng)計應用，對于FFT此類耗時大、數(shù)據(jù)量大的科學計算問題并不適用，因此需要對Flink相關的機制進行應用和改造，使得其符合FFT計算的要求。

在Flink流處理框架中設計實現(xiàn)良好的FFT算法并行化流程及優(yōu)化，不僅可以將FFT計算中的源數(shù)據(jù)流的特征表現(xiàn)得非常突出，推動了FFT計算在高精度、大寬帶和實時性的要求上更進一步；而且為傳統(tǒng)科學計算任務在大數(shù)據(jù)時代背景下，提供了一種基于分布式流處理平臺的并行流程新思路。

1 相關研究

1.1 單機FFT算法優(yōu)化現(xiàn)狀

目前在單機上進行FFT并行設計、提高FFT計算速度的方法主要有兩類。

一類主要優(yōu)化FFT算法本身。1995年Varkonyi-Koczy[6]提出了新的遞歸FFT算法，力圖提高處理時間。在生物醫(yī)學方面，Brünger[7]等人提出了基于覆蓋晶體晶胞的子網格的FFT變種算法，嘗試使用自網格減少計算的內存消耗。郭金鑫[8]等人重構蝶形網絡，在ARMV8及X86-64架構上構建了一個高性能FFT算法庫。FFT高性能程序庫、基于C的 FFTW[9]，通過高度優(yōu)化的多線程代碼和緩沖區(qū)優(yōu)化等手段，成為目前行業(yè)內頂尖的單機并行FFT程序庫。與此類似的包括基于GPU的cuFFT[10]和基于Java的Jtransform[11]等。

另一類主要在集中式的硬件平臺上進行并行化優(yōu)化。例如文獻[12]將多塊DSP芯片集成在一片信號處理器中，輔以多種硬件加速器，對一個FFT輸入數(shù)據(jù)進行計算。文獻[13]采用DSP芯片實現(xiàn)高速信號采集和FFT的實時運算。這種平臺已有成熟的產品，例如TMS320C6678多核DSP公司的雷達信號處理系統(tǒng)等。

但是由于單機的資源有限，單機和集中式環(huán)境擴展性不高，前端傳感器輸入數(shù)據(jù)量大、數(shù)據(jù)速率快，雖然已經有相當多的對FFT算法本身和對單機并行FFT程序庫的優(yōu)化研究，但是處理大規(guī)模數(shù)據(jù)仍然效率不高，且無法充分發(fā)揮集群機器CPU和內存資源在解決此類大規(guī)模問題上的優(yōu)勢[14]。因此使用大數(shù)據(jù)分布式框架成為目前FFT并行處理的重要手段。

1.2 分布式FFT算法并行流程及優(yōu)化現(xiàn)狀

目前已有很多在分布式框架上搭建FFT應用的研究工作。Duc[3]在 Hadoop和 Spark系統(tǒng)上設計了一種可擴展的FFT計算系統(tǒng)，以提供給聲學進行分析和使用。王菊[15]等人基于MapReduce改進了FFT算法，設計不同的組件按順序提交數(shù)據(jù)補零、變址運算、蝶式計算、格式化四個階段，使得以時間為基準的基-2 FFT算法能在框架上并行執(zhí)行，并應用于風電機組的發(fā)電場景上。趙鑫[16]等人設計了一種非遞歸的增量式FFT方法，在計算過程中加入隊列結構緩沖數(shù)據(jù)以解決數(shù)據(jù)傳輸過程中丟失和亂序問題，并應用于轉子合成軸心軌跡監(jiān)測中。

但目前FFT分布式設計大多本質都是批量處理，要求先存儲后計算，這種方法一方面有較多的時間花費在IO讀寫上，需要對大批量的數(shù)據(jù)進行存儲復制，另一方面無法實現(xiàn)實時的數(shù)據(jù)處理，數(shù)據(jù)延遲高。

由于單機環(huán)境和分布式環(huán)境有本質區(qū)別，單機FFT算法并未考慮其在分布式環(huán)境下的多機并行等問題。本文根據(jù) FFT算法本身的特點，設計了FFT在流式引擎Flink中的并行流程。將每幀數(shù)據(jù)并行拆分，在多級并行實例間構建蝶式變換，隨后逐級合并，同時在多幀數(shù)據(jù)間實現(xiàn)流水線計算。本文設計了蝶式計算遞歸深度的自動調優(yōu)，以自動設置在不同點數(shù)、不同集群中的最優(yōu)深度，將計算時間降至最低。另外，本文對Flink原有窗口進行了改造，實現(xiàn)以數(shù)據(jù)空間信息為觸發(fā)條件的窗口機制，使得蝶式計算數(shù)據(jù)全部到達時觸發(fā)窗口計算。同時在窗口內設計了多幀緩沖隊列，以防出現(xiàn)位置覆蓋、亂序、計算錯誤等問題。

2 方法設計

2.1 FFT流處理并行流程設計及優(yōu)化

本文參考了1965年Cooley和Tukey提出的經典遞歸基 2-FFT算法[17]，將其稍加改造，使得其適用于分布式流處理場景。

本文提出的基于Flink的FFT并行算法整體流程如圖1所示。該并行流程主要分為四個部分，分別為蝶式計算數(shù)據(jù)劃分、位置窗口觸發(fā)計算、小數(shù)據(jù)塊FFT計算和蝶式計算多級合并。

圖1 FFT并行算法整體流程圖

FFT數(shù)據(jù)的輸入流實質為每幀雷達數(shù)據(jù)中的每個點。蝶式計算數(shù)據(jù)劃分時，根據(jù)下游計算并行實例個數(shù)，將數(shù)據(jù)流按在整幀中的位置信息分發(fā)到對應的計算并行實例中，映射關系示例圖如圖2所示。某個位置的數(shù)據(jù)與計算并行實例編號的對應關系在初始化時計算并存于映射表中，在運行中可以直接從中取值，無需消耗額外的時間進行多次遞歸計算匹配。

圖2 基2-FFT點數(shù)為8時遞歸兩次后的位置映射圖

數(shù)據(jù)合并部分按蝶式計算遞歸合并的原則，將奇偶項劃分開并將單獨計算的兩塊數(shù)據(jù)塊重新合并。計算公式如式（1）所示。式中o為將要合并后的數(shù)組，u與v分別為上游處理偶數(shù)項和奇數(shù)項數(shù)據(jù)的并行實例的輸出數(shù)組，w為旋轉因子。

圖3為Flink中蝶式計算遞歸次數(shù)為2時FFT并行流程方法示意圖，Map算子主要負責將流元素進行位置編號，記錄一幀數(shù)據(jù)的最大點數(shù)和當前位置，將其位置標簽貼在每一個流動的數(shù)據(jù)項中。Partition算子負責將流元素實時拆分到不同的FFT計算算子中。在FFT計算算子中，符合要求的流數(shù)據(jù)全部到達后會觸發(fā)新設計的緩存窗口，對到達的數(shù)據(jù)直接進行FFT計算。Union算子負責將計算算子的輸出數(shù)據(jù)進行匯總合并。

圖3 蝶式計算遞歸深度為2時FFT并行流程示意圖

本文結合流式數(shù)據(jù)持續(xù)流動的特性，將流水線的并行思想應用于FFT并行流程方法中，實現(xiàn)處理時間重疊，從而降低每幀數(shù)據(jù)處理的時間。計算算子和同一深度的合并算子可以同時處理一幀的數(shù)據(jù)，而每一層深度之間可以流水線處理連續(xù)幀的數(shù)據(jù)。

由于集群性能不同和環(huán)境不同，為了讓分布式流處理中的FFT并行流程具備在各個不同集群環(huán)境主動適應、自動調優(yōu)的能力，需要通過試運行，確定FFT算法蝶式計算的最優(yōu)遞歸次數(shù)，從而將每幀的處理時間降至最低。

圖4是蝶式計算遞歸次數(shù)為1和2時FFT計算并行流程方法樣例。當深度為1時，F(xiàn)FT計算并行實例數(shù)量為2，將整個幀中的奇數(shù)項和偶數(shù)項分別交給第一個和第二個計算并行實例獨立計算。當深度為0時，整幀數(shù)據(jù)都交給一個并行實例計算，其本質上為Flink上的串行計算。深度最大為log（n），n為FFT點數(shù)。

圖4 深度為1和2時FFT并行流程示意圖

2.2 適用于FFT計算的緩存窗口設計

Flink現(xiàn)有窗口機制包括應用于事件時間的時間窗口、應用于流元素本身的計數(shù)窗口和應用于時間段數(shù)據(jù)活躍度的會話窗口，對于FFT此類高通量儀器的數(shù)據(jù)流來說，會在很短甚至相同的時間內產生大量的數(shù)據(jù)，這些雙精度浮點數(shù)的數(shù)據(jù)具有極其相近甚至相同的時間戳，因此無法借助事件時間戳信息的時間窗口。而又因為 FFT計算不允許數(shù)據(jù)亂序，因此無法保障數(shù)據(jù)有序到達計數(shù)窗口。

因此，需要針對FFT這類科學計算問題，在Flink上設計特殊的緩存窗口，使其能夠依據(jù)當前數(shù)據(jù)項在幀中的位置信息，在計算并行實例所要求的小塊數(shù)據(jù)全部到達后觸發(fā)窗口邏輯，將數(shù)據(jù)交給并行實例來進行計算。而對于負責合并的并行實例來說，由于其數(shù)據(jù)流來源為上游的兩個并行實例，輸出數(shù)據(jù)塊必須嚴格遵守位置順序以防亂序，也需要此類緩存窗口。

由于數(shù)據(jù)流可能流速過快，在負責合并的并行實例緩存窗口中，本文設有對多幀數(shù)據(jù)的緩沖隊列，能夠在一定程度上解決流速過快而產生如多幀數(shù)據(jù)覆蓋后計算錯誤等問題，同時也能夠在各并行實例計算速度不均衡的情況下提前將計算速度快的并行實例所產生的數(shù)據(jù)進行保存，從而節(jié)約部分通信時間。

本文緩存窗口實現(xiàn)方式如下，表1是其中設計的部分字段。

表1 適用于FFT的緩存窗口中部分字段設計

unionId表明了在當前這一深度中此緩存窗口所在并行實例的索引，此字段是對應所負責上一級兩個并行實例的索引，例如0號并行實例主要負責合并上一級0號和1號并行實例的數(shù)據(jù)塊。size字段表示負責合并的數(shù)據(jù)數(shù)量，在基2-FFT計算中，始終是兩兩合并，因此默認為2。isInit是判斷初始化的標簽。miniData是上一級并行實例所輸出的小數(shù)據(jù)塊。cacheQueue是多緩沖隊列，隊列內的每項數(shù)據(jù)以映射表的形式保存，映射表的鍵為所負責合并的上一級并行實例索引，映射表的值為上一級對應索引并行實例的輸出數(shù)據(jù)塊miniData。hasNum實時統(tǒng)計多緩沖隊列中頭部的映射表中已就緒的數(shù)據(jù)塊數(shù)量。

圖5是負責合并的并行實例中緩存窗口觸發(fā)樣例圖，該并行實例主要負責上游索引為0和1的并行實例的輸出數(shù)據(jù)塊。階段1時，窗口內多緩沖隊列為空。在階段2時，該并行實例接收上游輸出數(shù)據(jù)＜1，data＞。＜1，data＞中的 1，表明數(shù)據(jù)流中此數(shù)據(jù)項來自上游索引為1的并行實例，data是處理后的數(shù)據(jù)塊。將此data存儲于隊列的第一個映射表的鍵1所對應的值上。鍵0的值暫且空缺，表明未有完整匹配的數(shù)據(jù)到達。階段3時，接收到了一項數(shù)據(jù)項＜1，data＞，依然來源于上游索引為 1的并行實例，繼續(xù)緩存此數(shù)據(jù)塊。階段4時，接收來自上游索引為 0的并行實例所處理后的數(shù)據(jù)項＜0，data＞。此時隊列頭部映射表中的兩項數(shù)據(jù)項已全部出現(xiàn)，彈出隊列頭，并將彈出后的兩塊數(shù)據(jù)塊在此并行實例中進行合并。

圖5 合并算子中緩存窗口觸發(fā)樣例圖

3 實驗結果與分析

3.1 實驗配置與測試樣例

由于本文主要測試的是Flink中FFT算法的各并行度時間性能和容錯性能，應盡量避免帶寬等因素給本實驗帶來的影響，因此本文主要使用Flink的Standalone模式，一個節(jié)點中設置了28個slot給不同的并行實例運行任務。具體實驗節(jié)點配置情況如表2所示。

表2 實驗節(jié)點配置情況

本文使用隨機生成雙精度浮點數(shù)作為FFT的輸入數(shù)據(jù)，每組數(shù)據(jù)點數(shù)不同，均為1 000幀。本文實驗所用的Flink作業(yè)拓撲圖由一個Source數(shù)據(jù)源輸入算子、一個 Map算子、一個 Partition算子、多個計算算子、多個合并算子和一個Sink輸出算子組成。Source算子和Sink算子分別是Kafka的消費端和生產端。

3.2 實驗結果與分析

該實驗主要是測試FFT并行流程中處理每幀數(shù)據(jù)所需要的時間。在FFT并行流程中，多幀數(shù)據(jù)以流水線的形式在作業(yè)拓撲圖中進行計算，因此每幀數(shù)據(jù)處理時間的統(tǒng)計均是流水線時間。

實驗結果如表3所示，表中是對于不同點數(shù)的FFT數(shù)據(jù)在不同F(xiàn)FT計算并行度的情況下處理每幀數(shù)據(jù)的流水線時間。

表3 不同點數(shù)、不同并行度下處理每幀數(shù)據(jù)的流水線時間

表3中，按行觀察，即從同一個FFT數(shù)據(jù)點數(shù)來看，隨著并行度的逐漸增大，處理每幀數(shù)據(jù)的流水線時間先是逐漸減少，后來又逐漸增大。計算時間減少是由于隨著計算并行度的增加，并行的算法帶來了時間上的優(yōu)化。但是隨著并行度越來越大，F(xiàn)FT并行流程所自動形成的作業(yè)拓撲圖越加龐大，拉長了整個計算流程，作業(yè)拓撲圖中的各項并行實例、對應的窗口和流管道等組件越來越多，其本身的開銷占比越來越高，因此導致處理每幀數(shù)據(jù)的流水線時間有所提高。

按列觀察，即從不同的FFT數(shù)據(jù)點數(shù)來看，隨著點數(shù)的逐漸增大，處理每幀數(shù)據(jù)的流水線時間基本都會成倍增加，一方面因為FFT算法的復雜度是O（nlogn），在 FFT點數(shù)總量 n翻倍的情況下，計算時間也因此會同步增加；另一方面是由于數(shù)據(jù)量的提高，在整個并行流程中傳輸一幀數(shù)據(jù)所需要的時間大大增加，特別是各并行實例窗口的觸發(fā)延時和并行實例之間的傳輸也需要更多時間，增加了時間的損耗。

從整體來看，隨著點數(shù)的逐漸增大，時間消耗最小的最優(yōu)并行度先增大后減小。在點數(shù)為16 384時，并行度為2的情況下處理每幀數(shù)據(jù)的流水線時間最小，成為所有并行度下的最優(yōu)并行度；在點數(shù)分別為 65 536、131 072、262 144時，最優(yōu)的并行度向后推移為8；在點數(shù)更大時，最優(yōu)的并行度反而減小。并行度逐漸增大的原因為在點數(shù)較小的情況下，每幀的數(shù)據(jù)量較小，使用并行算法的必要性不高，因為并行算法會帶來一定的開銷，對時間的優(yōu)化不明顯。但在點數(shù)增加、每幀的數(shù)據(jù)量提高時，使用并行計算帶來的時間優(yōu)化越來越明顯，因此計算所需要的時間也越來越少。在點數(shù)增加到一定規(guī)模如1 048 576后，最優(yōu)并行度反而下降。本文探究最優(yōu)并行度變化的原因如下。

本文統(tǒng)計了并行度為1時不同點數(shù)下每幀數(shù)據(jù)完整的計算時間和通信時間，如表4所示，以及點數(shù)為131 072時不同并行度下每幀數(shù)據(jù)的計算時間和通信時間，如表5所示。此處總時間為整幀數(shù)據(jù)全部的時間，計算時間包括FFT計算時間、多級合并時間，通信時間包含序列化反序列化時間、流通道數(shù)據(jù)傳輸時間、并行實例輸入緩沖區(qū)和輸出緩沖區(qū)等待時間等。

如表4所示，在并行度不變、點數(shù)逐漸增加到巨大規(guī)模時，計算每幀數(shù)據(jù)所需的通信時間增加幅度遠大于FFT點數(shù)規(guī)模增加所導致的計算時間增加的幅度。因此表3中最佳并行度會隨著點數(shù)增加到一定程度時降低，是因為此時通信時間逐漸占據(jù)了主要部分。如表5所示，在點數(shù)不變的情況下，并行度逐漸增加會使一幀的總時間增加。其中并行度增加時，由于每個并行實例處理的數(shù)據(jù)減少，因此小份數(shù)據(jù)的計算時間會急劇降低，但是并行度增加時會拉長整個作業(yè)拓撲圖，F(xiàn)FT計算的深度會增加，帶來更多的通信階段，通信時間也會因此增加。因此需要試運行來自動決定不同F(xiàn)FT點數(shù)的最佳并行度，從而將每幀數(shù)據(jù)的流水線計算時間降至最低。

表4 并行度為1（Flink中串行）時不同點數(shù)每幀數(shù)據(jù)計算時間和通信時間

表5 點數(shù)為131 072時不同并行度每幀數(shù)據(jù)計算時間和通信時間

圖6所示是FFT不同點數(shù)下，F(xiàn)link中串行的計算時間和對應最優(yōu)并行度的計算時間對比圖。從實驗結果中可以看出，本文設計出的FFT并行流程在不同點數(shù)的最優(yōu)并行度下與Flink上串行的FFT算法相比，處理每幀數(shù)據(jù)的流水線時間大大降低。

圖6 Flink中串行計算時間和最優(yōu)并行度計算時間對比圖

4 結論

本文將FFT的計算特點與分布式流處理相結合，提出了一種基于Flink的FFT并行流程方法，同時在Flink中設計了適用于FFT的緩存窗口，使得通信時間和計算時間有部分時間重疊，加快FFT處理時間。實驗結果表明，本文在Flink上FFT并行流程中降低了每幀的處理時間，提高了計算效率。下一步的工作重點是對FFT并行流程中的流水線進行進一步的優(yōu)化。