DoFFT：一種基于分布式數據庫的快速傅里葉變換方法

季 朋，李 暉，陳 梅，戴震宇

(1.貴州大學計算機科學與技術學院,貴州 貴陽 550025;2.貴州省先進計算與醫療信息服務工程實驗室，貴州 貴陽 550025)

0 引 言

快速傅里葉變換(FFT)[1]算法是數字信號處理中的核心算法，在頻譜分析、數字通信、圖像處理中有著重要的應用?？焖俑道锶~變換在天文學方面也有大量應用，一個典型的應用場景是對脈沖星信號進行相干消色散處理[2]。脈沖星輻射的電磁波在傳播過程中，由于受到星際介質的干擾，不同頻率的電磁波經過星際介質后，會引起觀測到的脈沖星輪廓展寬，所以需要進行消色散處理。消色散處理通常使用相干消色散算法。相干消色散算法的基本原理[3]是將基帶脈沖星數字信號進行快速傅里葉變換，然后將頻域上各頻點的信號乘以星際介質函數CHIRP即頻率的響應函數，最終逆傅里葉變換到時域，從而實現將不同頻率成分的脈沖星型號對齊到某一頻點。傅里葉變換的快慢直接決定相關消色散的處理效率。

為了滿足海量信號數據的存儲與分析需求，本文采用Greenplum分布式數據庫[4-5]存儲脈沖星信號。在數據庫中對數據計算分析通常使用SQL語句，但SQL語句不便于實現復雜的算法，例如快速傅里葉變換等科學分析算法，而用戶自定義函數(UDF)可以滿足這一需求。本文實現了UDF形式的快速傅里葉變換算法，為了提高算法效率，對算法的執行做了進一步的優化。

在Greenplum數據庫集群中有多個segment實例，每個實例能夠并行執行一部分任務。文獻[6]設計了并行結構的FFT算法，本文同樣對FFT進行了并行化設計，使每個實例能夠并行計算并得到一部分結果，從而提高算法執行效率。另一方面，集群中數據分布[7]也可能影響算法性能。Greenplum將數據分布到各個節點中，當在某個節點上執行UDF時，由于節點的負載等不同，會導致不同的性能。為了使UDF算法執行性能達到最優，本文做數據重分布。文獻[8]根據網絡傳播元組數目，重分布節點元組，從而提高表連接性能。本文是根據當前節點的數據分片大小、負載等因素重分布元組數據，能夠較大地提高FFT的執行效率。

1 預備知識

1.1 Greenplum分布式數據庫

分布式數據庫[9-10]是通過網絡將物理上分散的多個數據庫單元連接起來組成的一個邏輯上統一的數據庫。每個數據庫單元相當于一個數據庫實例，它存儲數據且并行執行查詢。在查詢時[11]，集群中的各數據庫實例能夠并行執行一部分查詢任務，從而提高查詢效率。

圖1 Greenplum架構

Greenplum是基于PostgreSQL 8.X的MPP分布式數據庫。MPP[12](shared nothing架構)指有2個或更多個處理器協同執行一個操作的系統。每個處理器都有自己的內存、操作系統和磁盤。圖1是Geenplum的架構圖。

Greenplum集群中的節點分為2類，一類是master節點，一類是node節點，node節點為物理節點，每一個node節點上都有多個數據庫實例segment，負責存儲與查詢數據。當一個查詢到來時，master節點上的優化器(optimizer)負責生成查詢計劃，而查詢調度器(QD)會以slice的形式下發查詢任務到每個數據節點，數據節點收到查詢任務后，創建查詢執行器(QE)進程執行任務。Greenplum將一個查詢計劃切分成多個slice來執行，多個查詢計劃并行執行。本文的UDF通過QE執行。

1.2 UDF技術

在編寫應用程序的時候，傳統的方法是讓服務器應用程序的處理邏輯盡可能地被推送到客戶端,而程序處理的數據也需要通過外部網絡從數據庫中讀取然后推送到客戶端。本文的方法是把計算、數據操作全部轉移到一個位于服務器上的UDF里面，然后直接在數據庫服務器中執行該UDF。UDF不僅便于設計更加復雜的算法，同時還消除了數據在客戶端與服務器之間傳輸的過程，減少了算法執行的時間。

Greenplum中的UDF能夠提供強大的擴展能力，比如能夠添加數據類型、創建新的索引等。用戶可以使用UDF實現內置函數不提供的功能和完成復雜算法。UDF可以用多種語言實現，比如Python、Perl等一些插件式的腳本語言。但腳本語言在某些情況下存在缺陷，一方面，當執行相同的算法時，大多數腳本語言會比C代碼運行得慢；另一方面，當遇到函數要被每個查詢調用成千上萬次的情況，那么運行過程會出現各種超負荷運作，如函數調用、參數與返回值在腳本語言間的轉化等?；诖?，本文采用C語言實現UDF形式的快速傅里葉變換算法。

2 數據重分布

Greenplum在進行數據分布[13]時是根據表的分布鍵按照分布策略把表數據分散到不同的segment實例。常見的分布策略包括哈希分布和隨機分布。哈希分布[14]是根據指定的分布列計算該列對應的每一行數據的哈希值，并映射行數據至相應的segment實例。隨機分布指記錄隨機分散數據，相同的記錄可能分布到不同的segment。本文采用的數據分布策略為哈希分布。但哈希分布僅考慮了數據的分布存儲，沒有考慮分布對查詢的影響。例如當一個節點的CPU利用率很高時，位于該節點上的segment的查詢執行效率會極大地下降?；诖?，在執行查詢前，本文對數據重分布，即根據當前各segment的負載和數據分布情況對表記錄重新調整分散，從而加快后續查詢的執行速度。

2.1 數據重分布方案

在調整數據分布前，需要確定當前數據分布的狀態，一個表的數據分布通常有如圖2所示的幾種情況。

圖2 數據分片

在圖2中，node1、node2、node3、node4為Greenplum的4個物理節點，與圖1的node1和node2類似。seg0，seg1，…，seg15為16個segment，與圖1中的segment類似，圖2中的a1～a16、b1～b4、c1～c4、d1～d3分別為表a、b、c、d的數據分片，例如表b有4個數據分片b1、b2、b3、b4，分別分布在seg2、seg6、seg10、seg14。數據分布分為以下2種情況：

1)數據均勻分布在各個segment上。

數據表a的記錄分布在集群中的所有segment中，這種情況不做處理，直接運行查詢。

2)數據分布在多個segment上。

如圖2中的表b、c、d，數據分布在16個segment中的少數幾個segment中。例如表c的數據分布在seg4、seg5、seg8和seg12，這種情況需要根據代價判斷是否需要做數據重分布。以數據表d為例，表d的數據分片為d1、d2、d3，分別位于seg4、seg8、seg12上。首先分析數據分片d1，d1位于seg4上，則需要計算d1均勻分布到除seg8、seg12外的其他14個segment的代價，一共有n種情況：

(1)

2.2 計算代價

C=負載+時間

(2)

代價可根據節點當前的負載[15]和預計所需時間得到，比如在計算I/O代價時候，需要評估節點的I/O負載，還需要估算數據從磁盤讀出和寫入磁盤的時間，然后根據這2個維度的值得出代價。

2.2.1 計算CPU代價

計算CPU代價時，首先需要得到系統當前的負載，負載越高，代價越大，估算負載的公式如下：

(3)

公式(3)對CPU當前的負載做歸一化處理，LCPU表示系統當前的負載，利用uptimie命令得到，φmin表示系統的最小負載，φmax表示系統的最大負載，由于本文的實驗環境中的機器的CPU為4核，所以φmax一般大于4。由于φmin可以為0，所以能夠得到公式(3)的右半部分。

在得到CPU的負載后，還需要估算CPU處理記錄的代價，計算公式如下：

(4)

公式(4)做歸一化處理，α為當前segment的記錄條數。βmin表示segment的最小記錄條數，βmax表示segment的最大記錄條數，對于d1來說，這個值為2個segment上存儲的記錄總數，同樣由于βmin可以為0，所以，能夠推導出等式的右半部分。

在得到負載和耗時的值之后，便可以計算CPU的代價：

(5)

由于負載對CPU代價影響較大，所以負載乘以100。從上述公式可以看出CPU的負載越大，CPU代價越大，同樣待處理的記錄條數越多，CPU的代價也越大。

2.2.2 計算I/O代價

計算I/O代價同樣需要得到負載和時間2個維度的值。負載能夠直接通過iostat命令得到，磁盤的使用率越高表示I/O的負載越大，I/O的代價LIO也就越高。假設從磁盤讀取的數據量為γr，向磁盤寫入的數據量為γw，讀磁盤的速率為δr，寫磁盤的速率為δw，則可得到代價為：

(6)

在得到2個維度的代價以后，則可計算I/O的代價：

(7)

上述公式中的100，同樣是為擴大負載對代價的影響。公式(7)表示當負載越大，讀寫磁盤的數據越多，I/O代價越大，反之，越小。

2.2.3 計算網絡代價

計算網絡代價與計算I/O代價類似，首先通過命令netstat得到網絡負載的代價Lnet，接著計算通過網絡傳輸數據的代價。假設傳輸的數據量為n×m kB，網絡的帶寬為ξ，對于傳輸數據的代價為：

(8)

由此得到網絡總代價為：

(9)

可見，網絡負載越高，傳輸的數據量越多，則網絡代價越大。

2.2.4 計算平均等待時間

在得到了CPU、I/O和網絡的代價，總代價為上述3個代價之和，如果對于2種不同的方案，總代價C相同或者相差不大時，還需要判斷每個segment的任務平均等待時間Tavg。該值通過公式(10)計算得到：

Tavg=(Q1+Q2+Q3+…+QZ)/z

(10)

公式(10)中的Q1，Q2，…，Qz為segment中在等待隊列中的查詢的等待時間，而z表示等待任務的個數。

2.2.5 計算總代價

一種方案的總代價為：

C=CCPU+CIO+Cnet

(11)

總代價即為CPU、I/O和網絡代價之和。在得到每種方案的總代價以后，比較所有方案的代價，采用代價最小的方案。如果待比較的2種方案的代價之差小于σ，則比較2種方案中待遷移到的segment上任務的平均等待時間Tavg，其中σ通過實驗根據經驗得到。下面是對于方案τ1和τ2，判斷采用哪種方案的算法，C1、C2表示2種方案的代價，T1、T2表示2種方案中segment的總平均等待時間。算法1為選擇方案算法。

算法1選擇方案算法

if(C1=C2or abs(C1-C2<σ){

if(T1

use τ1

}else{

use τ2

}

}else if(C1

use τ1

}else{

use τ2

}

上述算法能夠得到一種代價最小或代價與平均等待時間都較小的方案，該方案能夠最大程度地提高快速傅里葉變換執行效率。

3 快速傅里葉變換的并行實現

根據第2章得到的數據分布方案對數據進行重分布后，還需要設計并行化的快速傅里葉變換算法。

3.1 快速傅里葉變換

快速傅里葉變換是對離散傅里葉變換(DFT)的改進，DFT的計算過程如下。

給定向量A=(a0,a1,…,an-1)，DFT將A變換為B=(b0,b1,…,bn-1)，變換的過程為：

(12)

式(12)中的w=e2πi/n，式(12)表示的含義寫成矩陣形式為：

DFT的另外一種形式為：

(13)

舉一個具體的例子：對于一個4點的向量X=[2,3,3,2]，DFT的計算過程如下：

圖3 FFT算法

從圖3可以看到，對于一個初始向量X[n]，可以分段得到結果x[k]，因此，可以通過在Greenplum中的各個segment并行執行得到部分結果，然后在master上匯總結果并返回。

3.2 分布式并行執行

根據快速傅里葉變換的算法原理，能夠設計多進程并行執行的FFT[16-18]。由于數據存儲在Greenplum中，可以利用Greenplum的分布式特性，通過讓多個segment并行運行來實現多進程執行的效果。利用Greenplum的UDF實現快速傅里葉變換的算法設計見算法2。

算法2快速傅里葉變換的分布式并行執行

master:

get_situation();

for(i=0;i

{

w[i].r=cos(i*2*PI/wLength);

w[i].i=sin(i*2*PI/wLength);

}

parallel segment:

cal_partial();

send_partial();

master:

recv_gather();

上述算法共分為3個步驟：

1)在master上執行準備工作，例如得到數據分布情況、初始化旋轉因子數組等。

2)在segment上執行快速傅里葉變換。

3)在master節點上匯總各并行的segment結果并返回給客戶端。

算法的主要函數說明如下：

1)get_situation()。得到數據的分布情況，包括：數據分布在幾個segment上，每個segment有多少條記錄等。在得到數據的分布情況后，接著初始化旋轉因子數組，并把該數組發送給各segment。

2)cal_partial()。每個segment執行具體的快速傅里葉變換，該函數能夠得到最終結果的一部分。

3)send_partial()。每個segment得到一部分結果以后，把結果送給master節點。

4)recv_gather()。接收各segment的結果并匯總，返回給客戶端。

4 實驗及結果分析

本文工作中采用的Greenplum版本為5.0.0-alpha+79a3598。集群共有5個節點，1個主節點和4個從節點，從節點主要用來存放數據并執行查詢，主節點則負責分配查詢和匯總結果。主節點的硬件配置為32 GB的內存，CPU為4核Intel(R) Xeon(R) CPU E5-2630 v2@2.60 GHz，從節點的內存16 GB，CPU的核數和型號與主節點相同，在每個從節點中有4個數據庫實例。主節點和從節點的操作系統均為CentOS 7.4，Linux內核版本為3.10。

本文的測試數據為用Python模擬生成的脈沖信號，數據量最大為10 GB，在數據庫中使用列存儲方式存儲數據。使用列存儲有利于數據壓縮，并且能減少查詢的數據量。

4.1 并行化執行對效率的影響

為了驗證DoFFT算法并行執行的效果，實驗比較當所有數據存儲于集群中的一個segment中與均勻存儲于所有segment時的FFT執行效率。數據量范圍為1 GB～10 GB。性能結果如圖4所示。

圖4 并行FFT效率對比

從圖4可以看出，DoFFT算法能夠極大提升FFT的效率，并且隨著數據量的增多，提升效果也更明顯。

4.2 執行環境對效率的影響

上節比較了DoFFT并行化算法對FFT效率的提升，本節比較DoFFT對于不同的環境負載的優化效果。

4.2.1 CPU

CPU對執行效率的影響主要在于，若某個segment的CPU負載很高，則會降低該segment上FFT的執行效率。本實驗中創建一個高負載的環境，使CPU的平均負載高于4，對比該執行環境下，直接執行FFT所用時間與使用DoFFT算法后的時間，如圖5所示。

圖5 CPU影響

從圖5可以看出，在每個數據量級別下，DoFFT算法對FFT性能都有2倍左右的性能提升。

4.2.2 I/O

I/O對FFT執行效率的影響主要來自于節點的I/O負載和I/O帶寬。由于集群中的節點具有相近的I/O帶寬，所以主要比較DoFFT算法對I/O負載的優化效果。結果如圖6所示。

圖6 I/O影響

從圖6可以看出，在小數據量時，DoFFT算法提升效果并不明顯，但隨著數據量的增加，效果越來越明顯。

4.2.3 網絡

Greenplum在執行FFT時，數據會在網絡中傳輸，網絡的負載和帶寬對性能會產生一定影響。圖7展示了DoFFT基于網絡因素對FFT的優化效果。

圖7 網絡影響

從圖7可以看出，DoFFT算法能夠小幅度地提高性能，但較CPU和I/O，對網路的提升效果不明顯。

4.2.4 平均等待時間

分布式數據庫中的segment執行的任務數會不同，任務平均等待時間也會不同，平均等待時間越長，處于等待隊列中的查詢被調度執行所需的時間也就越長。圖8展示了當某些segment中執行較多任務時，DoFFT的優化效果。

圖8 平均等待時間影響

從圖8可以看到，平均等待時間同樣能夠小幅地提升效率。

4.2.5 實驗總結

上述的實驗結果表明，DoFFT算法能夠不同程度提高算法的執行效率。DoFFT的并行化對FFT執行提升效果最明顯。在各影響因素中，對CPU和I/O優化的的提升效果較明顯，對網絡和平均等待時間優化的提升較小。

5 結束語

為提高快速傅里葉變換算法在分布式數據庫中執行的效率，本文提出了DoFFT算法。算法首先對Greenplum中的數據進行重分布。Greenplum在存儲數據時，會使用哈希分布或隨機分布策略分散表數據至各個segment實例，而在實際環境中，每個segment的CPU、I/O等負載不同，硬件配置環境不同，會造成UDF形式的FFT算法在執行時達不到最優性能。DoFFT算法以每個segment的負載、處理時間等量化為代價，采用一種代價最小的方案對數據做重分布。并且，為進一步提升FFT執行效率，DoFFT算法還利用Greenplum中各個segment實例并行地執行FFT。實驗結果表明，DoFFT算法能夠較好地提升FFT在分布式數據庫中執行的性能。

參考文獻：

[1] Cochran W, Cooley J, Favin D, et al. What is the fast Fourier transform?[J]. Proceedings of the IEEE, 1967,15(10):1664-1674.

[2] 劉東亮, Demorest P, 南仁東. 基于CUDA的相干消色散算法實現與測試[J]. 科學技術與工程, 2010,10(8):1947-1950.

[3] 徐永華,李紀云,張穎倩,等. 相干消色散脈沖星觀測系統的研究[J]. 天文研究與技術, 2015,12(4):480-486.

[4] Caragea G C, Garciaalvarado C, Petropoulos M, et al. Total operator state recall—Cost-effective reuse of results in Greenplum Database[C]// IEEE International Conference on Data Engineering Workshops. 2013:44-49.

[5] Antova L, El-Helw A, Soliman M A, et al. Optimizing queries over partitioned tables in MPP systems[C]// Proceedings of 2014 ACM SIGMOD International Conference on Management of Data. 2014:373-384.

[6] 丁順英. 基于并行結構的FFT算法的軟硬件設計與實現[D]. 哈爾濱:哈爾濱工業大學, 2012.

[7] Cheng Long, Li Tao. Efficient data redistribution to speedup big data analytics in large systems[C]// IEEE International Conference on High Performance Computing. 2017:91-100.

[8] Polychroniou O, Sen R, Ross K A. Track join: Distributed joins with minimal network traffic[C]// Proceedings of 2014 ACM SIGMOD International Conference on Management of Data. 20141483-1494.

[9] ?zsu M T, Valduriez P. Distributed database systems: Where are we now?[J]. Computer, 2002,24(8):68-78.

[10] Rothnie J B, Goodman N. A survey of research and development in distributed database management[C]// International Conference on Very Large Data Bases.1977:48-62.

[11] Yu C T, Chang C C. Distributed query processing[J]. ACM Computing Surveys, 1989,16(4):399-433.

[12] Dewitt D J, Gray J. Parallel database systems:The future of database processing or a passing fad?[J]. ACM Sigmod Record, 1990,19(4):104-112.

[13] Garcia-Alvarado C, Raghavan V, Narayanan S, et al. Automatic data placement in MPP databases[C]// IEEE International Conference on Data Engineering Workshops. 2012:322-327.

[14] Shen Yuxia. Distributed storage system model design in Internet of things based on hash distribution[J]. International Journal of Security & Networks, 2017,12(3):141..

[15] Curino C, Jones E P C, Madden S, et al. Workload-aware database monitoring and consolidation[C]// ACM SIGMOD International Conference on Management of Data. 2011:313-324.

[16] Gro?e P, May N, Lehner W. A Study of Partitioning and Parallel UDF Execution with the SAP HANA Database[EB/OL]. https://www.researchgate.net/publication/266660040_A_study_of_partitioning_and_parallel_UDF_execution_with_the_SAP_HANA_database, 2018-03-13.

[17] Taboada J M, Araujo M G, Basteiro F O, et al. MLFMA-FFT parallel algorithm for the solution of extremely large problems in electromagnetics[J]. Progress in Electromagnetics Research. 2010,101(2):350-363.

[18] Chu E, George A. FFT algorithms and their adaptation to parallel processing[J]. Linear Algebra & Its Applications, 1998,284(1):95-124.