基于大數據技術的Cochran-Armitage算法的分布式研究與實現

吳雙列　王軍凱　杜江　高洪睿

關鍵詞：趨勢檢驗算法；分布式；并行計算；大數據技術；計算集群

中圖分類號：TP311.11 文獻標識碼：A

文章編號：1009-3044（2024）03-0078-07

0 引言

在基因組分析中，Cochran-Armitage趨勢檢驗是非常重要的算法，其主要被用來完成基因型頻率差異的統計檢驗[1-2]。Cochran-Armitage趨勢檢驗在實際基因檢測時，卻存在兩個問題：第一，每個生物體有大量基因信息，使用傳統的Cochran-Armitage趨勢檢驗算法進行計算，花費的時間成本會比較高。例如，一個人類樣本可能包含超過4 000 000個染色體。如果有兩個組，第一組有3 000個樣本，第二組有5 000個樣本。若要去比較兩個組的基因型頻率，必須分析32 000 000 000條記錄。第二，Cochran-Armitage趨勢檢驗算法每次只能對兩個組進行計算，若同時有多個組要處理，則除正在運行的任務外，其余的處理任務可能因需要等待計算資源而無法及時處理。

針對以上兩個問題，本文提出了基于大數據技術的分布式并行化Cochran-Armitage算法[3-8]。首先，本文使用分布式文件存儲系統HDFS存儲需要計算的數據并設計文件的讀取邏輯。其次，設計算法的Map階段以及Reduce階段的處理邏輯；最后，實現了Spark 計算集群執行分布式并行化的計算任務。這種分布式的Cochran-Armitage趨勢檢驗算法，不僅極大地縮減了運算時間，同時也降低了計算機資源的占用。

1 研究現狀

Cochran-Armitage趨勢檢驗算法常被應用在小樣本和小批量的實驗中。針對趨勢檢驗的改進有兩種方法：第一種是用其他的模型替換Cocharan-Armitage。例如2023年，Manning和Ku等人在遺傳病例對照關聯研究中使用了Jonchheere-Terpstra趨勢檢驗[9]，這是Cochran-Armitage趨勢檢驗的非參數替代方法。第二種是將趨勢檢驗與其他模型結合。例如2023年，Mesa和Analuisa等人在研究肉瘤患者合并癥與住院死亡率之間的關聯時將Cochran-Armitage趨勢檢驗與多元二項式邏輯回歸結合[10]。

對于大樣本和大批量的應用場景，它們的效果并不太理想。傳統的趨勢檢驗算法都是串行實現，會消耗大量的時間并降低了工作效率。所以，文本對分布式并行化的Cochran-Armitage 算法展開了深入的研究。

2 Cochran-Armitage 分布式實現

2.1 Cochran-Armitage 趨勢檢驗原理

20世紀下半葉，Cochran和Armitage提出并完善了Cochran-Armitage 算法[11-12]，Cochran-Armitage 算法一經推出大受好評，被認為是基因科學中非常重要的一個計算指標。在基因組分析中，Cochran-Armitage 算法可以完成基因型頻率（genotype frequency） 差異的統計檢驗，主要用來計算P值，P值越小則差異越大。Cochran-Armitage趨勢檢驗根據基因數據為每個等位基因建立2行3列的列聯表。列聯表的行表示分組，列聯表的列表示該等位基因在每條數據中出現相應計數的條數。

在表1和表2中，行標簽（組A與組B） 表示基因組分析的兩個不同組別，兩個組的樣本都是不相同的。列標簽（0的計數、1的計數和2的計數）表示在某個組中一條基因數據里某個等位基因存在0、1或2個的樣本數量。例如，組A有2個樣本，其中一個樣本的等位基因1為A，等位基因2為C；另一個樣本的等位基因1 為A，等位基因2為A。則對于等位基因A的列聯表中，組A中0的計數表示為0；1的計數表示為1；2的計數表示為1。表2中位于行標簽和列標簽的總和分別表示對某一行求和與對某一列求和。

得到列聯表后，分別計算行和列的邊緣總和，以及整個表的總和。

2.2 經典串行Cochran-Armitage 算法的分析

串行算法是單臺機器單線程的處理任務。流程如圖1。

如圖1 所示。首先，初始化分布式文件系統HDFS既獲取與分布式文件系統HDFS的連接，為讀取數據做準備。接著，獲取組A 與組B 數據文件（aGroup.txt 和bGroup.txt） 的輸入流并更新列聯表。最后，分別計算各個等位基因的P 值并輸出結果。

1） 存在的問題

①問題1：圖1中每個矩形都代表一個邏輯任務。對于串行計算，每一個任務都要等待上一個任務徹底的處理完成才可以開始。但是這些任務之間并不是完全依賴關系，對于獲取aGroup.txt輸入流讀入數據切割并更新列聯表的階段和獲取bGroup.txt輸入流讀取數據切割并更新列聯表的階段，這是兩個完全不依賴彼此執行的階段?？蓪蓚€階段并行執行來解決，但串行任務處理的特性限制了這個方法。

②問題2：在讀入數據切割并更新列聯表階段，對于處理大數據量問題，如百萬數據、千萬條甚至億條數據，都是單機處理。如果一條數據處理用時0.05ms，那么1億條數據就需要消耗5 000 000ms，這是非常大的資源消耗。如果將這些數據交由N臺機器同時處理則可解決這個問題。每臺機器所處理的數據量以及處理數據所消耗的時間均減少至原來的1/N倍。

上述2個問題在單機串行模式下無法解決，但可以在并行分布式模式下被解決。

2.3 并行的Cochran-Armitage 算法的設計

1） 串行算法問題的解決并行分布式模式可以解決上述串行算法存在的問題，并且執行效率更高，對資源的消耗更少。

①分布式并行化方法1：將數據的讀取分別交由多臺機器，一些機器讀取aGroup.txt數據，一些機器讀取bGroup.txt數據，即可在同一時間開啟多個任務并發執行。

②分布式并行化方法2：每臺機器都會有自己讀取的部分數據，根據計算向數據靠近的原則，在讀取數據成功的機器上啟動計算邏輯處理數據。這實現了分布式集群計算。

2） 分布式流程

并行分布式整體運行流程如圖2所示。首先，主節點提交應用程序主體，為應用程序建立基礎運行環境Context。Context負責與集群管理器通信，以及進行資源的申請、任務的分配和監控等。接著，集群管理器為執行器分配資源，啟動執行器進程。執行器運行情況會隨著“心跳”發送到集群管理器上。然后，Con?text根據程序的依賴關系構建有向無環圖（DAG） ，將DGA提交給DGA調度器解析。DAG被切分成多個階段（stage，每個stage都是一個任務集），Context計算出各個stage之間的依賴關系后將任務集提交給任務調度器進行處理。執行器向Context申請任務后，任務調度器將任務分發給執行器運行。同時，Context將應用程序代碼發放給執行器。最后，執行器執行任務，將執行結果反饋給任務調度器和DAG調度器。任務執行完畢后寫出數據并釋放資源。

圖2在邏輯上描述了整個分布式執行流程，包括使用的組件以及各個組件之間的通信與數據傳遞關系。同時描述了工作節點所需要執行的任務以及任務之間的聯系。每個工作節點都有自己的執行器進程，目的是當工作節點執行多個應用程序時，將應用程序彼此隔離。

3 并行分布式Cochran-Armitage 算法的實現

本文將Cochran-Armitage趨勢檢驗算法與大數據分布式計算框架Spark相結合。Spark分布式計算框架可將算法整體分割成數個小任務單獨運算，并將每個小任務單獨運算的結果整合。Spark借助分布式文件系統HDFS存儲數據文件，分布式文件系統HDFS 可將大體量數據文件切割成塊（Block） 。這可在處理大數據文件中實現最小化尋址開銷。

3.1 分布式集群的基本架構

普通的文件系統只需要單個計算機節點（由處理器、內存、高速緩存和本地磁盤構成）就可以完成文件的存儲和處理。分布式文件系統把文件分布存儲到多個計算機節點上，成千上萬的計算機節點構成計算機集群。目前的分布式文件系統所采用的計算機集群都是由普通硬件構成，這大大降低了硬件上的開銷。集群中的計算機節點存放在機架上，每個機架可以存放多個節點，同一機架上的不同節點之間常通過以太網互聯，多個不同機架之間采用網絡或交換機互聯。本文描述的Data Node和Name Node均為機架中的單一節點。分布式集群的基本架構如圖3所示。

3.2 分布式存儲數據

本文數據都存儲在分布式文件系統HDFS中。存儲過程如圖4。

如圖4所示，被Cochran-Armitage 算法處理的基因文件通過HDFS客戶端上傳，具體分如下4步：① HDFS客戶端（HDFS Client） 獲取文件信息，將文件切成多個塊（Block） 。每一個Block的大小默認是128M。切分完成后，Client保存文件的切片信息，用于下一步的文件上傳工作。

② HDFS Client創建Distributed File System對象。Distributed File System對象負責與Name Node節點通信，主要工作有以下兩個：工作1是將文件上傳請求發送給Name Node，并接受Name Node對文件上傳請求的響應；工作2是向Name Node請求獲取Data Node，并接受Data Node節點信息。Data Node存儲上傳文件。

③ 當Name Node響應Data Node的信息后，Client 會與相應的Data Node通信。

④ HDFS Client 收到Data Node 的應答后開始上傳數據塊。HDFS Client創建FSData Output Stream對象開始文件傳輸。

3.3 分布式讀取數據

執行分布式并行計算所需要的數據均存儲在分布式文件系統HDFS。所以在開始計算之前需要讀取數據。具體讀取過程如圖5所示。

① HDFS Client 創建Distributed File System 對象負責與Name Node進行通信。Name Node返回文件塊Block所在Data Node的元數據信息。其中包括Block 大小和Data Node信息等。

② HDFS Client創建FSData Input Stream對象，并根據Name Node返回的Data Node信息向Data Node發送讀取文件塊請求。

③ Data Node接收請求后，開始將文件以字節流的形式發送給HDFS Client。

3.4 分布式并行計算的Map階段

Map階段是分布式并行計算的第一步，其核心處理邏輯在map 函數。所有經過圖5 中所示的HDFSClient讀入的數據都會以鍵值對形式交給map 函數處理。map函數會對基因數據做轉換，并將轉換后的數據傳輸到Reduce 階段處理。Map 階段流程如圖6。

① Record Reader對象組合Input Format對象，將基因數據塊Block以鍵值對形式從HDFS讀入。鍵值對中的鍵為數據的偏移量，鍵值對中的值為讀取到的一行數據。

② Map Task（Map 任務）中由map函數處理數據，具體過程如圖7所示。

③ Map Task（Map任務）生成的中間數據會根據Hash 分區算法放到對應的bucket（桶）中。每一個Map Task根據Reduce Task（Reduce任務）的數量創建相應的bucket。bucket的數量為m × r 個。其中m 是Map Task的個數，r 是Reduce Task的個數。

④ Bucket中的臨時數據會在Reduce階段被再次抓取和處理。

Map階段主要依靠map函數處理數據。map函數會對基因數據做轉換，并將轉換后的數據傳輸到Re?duce階段處理。被切分后的基因數據塊通過HDFSClient從HDFS讀入Spark計算框架?；驍祿K中的所有數據需要被Map階段中map函數處理?；驍祿枣I值對的形式讀入。其中鍵K為數據在文件中的偏移量，值V為文件中的一行數據。在Spark 以圖5方式讀取數據塊的過程中，對數據做了預處理，刪除了K，只保留了V。map函數的具體實現邏輯如圖7所示。

① 預處理后數據格式為chromosome-ID：chromosome-start-position：chromoso-me-stop-position;GROUP-NAME|allele1|allele2。其中chromosome-ID 表示染色體號；chromosome-start-position 和chromosome-stop-position 表示基因的起止位置；GROUP-NAME 表示組別編號；allele1 和allele2 分別表示等位基因。

②切割字符串獲取等位基因allele1和allele2并更新列聯表。Cochran-Armitage算法是對2行3列的列聯表做計算。列聯表為二維。這里將列聯表中每組的數據表示成長度為15的一維表。因為每個等位基因對應一個2行3列的列聯表，列聯表的行表示分組。本文用到的數據集有5個等位基因，2個組別。所以每個組中一個等位基因要使用3個數據位，5個等位基因使用15個數據位。

③因為更新的一維列聯表只能屬于一個分組，所以要為列聯表標記其屬于哪個分組。標記后產生的結果以及中間文件的數據分兩部分，一部分表示分組，另一部分表示一維列聯表。

3.5 分布式并行計算的Reduce 階段

Reduce階段中的數據來自它的上一個Map階段生成并存儲在bucket的中間文件。文件中的數據均由Reduce Task中的reduce函數處理。被處理后的數據會被拉取到主節點進一步處理。具體的Reduce階段流程如圖8所示。

① Map階段處理后生成的數據文件會有多個分區存儲在bucket中。每個bucket對應一個Aggregator。Aggregator 對數據做歸并排序操作。Aggregator 本質上是一個HashMap對象。里面的元素為鍵值對 形式。Aggregator 拉取一個數據，若是HashMap不存在當前key，則插入數據；否則，把value的值累加到V上。

② Reduce Task拉取自己對應分區的數據并交由reduce函數處理。reduce函數處理流程如圖9 所示。Reduce Task拉取數據，其中k是分組，v是長度為15的一維表。流程如下：

reduce函數是按照數據的鍵進行分組，對于鍵相同的數據每次處理兩個。例如：

數據1：

數據2：

對數據1的V與數據2的V在相同索引位相加，得到結果數據：：

Reduce階段完成后，將產生2條鍵值對數據。一條是。另一條是。

3.6 分布式并行化的趨勢檢驗的實現

趨勢檢驗需要2行3列的二維列聯表。因為Re?duce階段生成的數據是長度為15的一維表，所以需要在邏輯上將一維表切分為5份。每一份代表一個等位基因在對應二維列聯表中的一組的數據。圖10給出了利用趨勢檢驗算法計算結果并保存的過程：

4 實驗結果與分析

4.1 實驗數據集

實驗所需要的基因數據參考千人基因組計劃中的vcf文件[15]生成。本文共使用4組實驗數據，每組兩個文件。每個文件包含的數據量為：4 946 865條基因數據，12 367 134條基因數據，24 734 269條基因數據和74 202 843條基因數據。每一條均為下列格式的數據：chromosome-ID：chromosome-start-position：chromosomestop-position;GROUP-NAME|allele1|allele2。其中chromosome-ID 表示染色體號；chromosome-startposition和chromosome-stop-position 表示基因在染色體中的起止位置；GROUP-NAME表示分組；allele1和allele2均表示等位基因。chromosome-ID：chromosomestart-position：chromosome-stop-position在計算中用于將同一染色體和同一起止位置的數據分組。GROUP-NAME 用于數據按組別分組。allele1 和al?lele2用于構建列聯表。

例：21：9411239：9411239;b|A|A

4.2 實驗環境

實驗環境分為單機串行和分布式并行。單機串行環境下，計算機系統為Ubuntu18.04，處理器為i7-8750H，內存16G， Java版本為Open JDK 1.8。分布式并行環境下，一共有3臺機器，其職責如圖2所示。一臺機器充當主節點和工作節點的職責，另外兩臺充當工作節點的職責。3臺機器的硬件條件均相同，其計算機系統為Ubuntu18.04，處理器為i7-8750H ，內存16G， Java版本為Open JDK 1.8，Hadoop版本為3.1.3，Spark版本為3.3.0。

4.3 評價指標

本文選擇了4個評價指標。（1） 機器數：單機串行環境只有1臺機器，分布式并行環境則有大于等于2 臺的機器。以不同的機器數量執行任務，所耗費時間一定不同。本文假設執行任務消耗的時間隨機器數的增加而減少。（2） 數據量：分布式并行環境應該更能勝任大數據量的情況。數據量越大，串行計算和分布式并行計算的性能差距就越明顯。（3） CPU內核使用率和內存利用率：CPU用于執行計算，內存用于交換數據。執行計算機任務應是對資源的利用越合理越好。CPU內核使用越多，則計算效率越高。內存使用越少，資源浪費越少。

4.4 實驗結果與分析

① 機器數與任務執行時間的關系

只使用1臺機器則屬于單機串行環境。機器數大于1臺且所有機器執行同一任務則屬于分布式并行環境。本文分別使用了1臺、2臺和3臺機器執行趨勢檢驗任務，每一次執行的任務和使用數據集均相同。不同情況下的時間花費如圖11所示。相同數據集相同任務情況下，執行時間隨著機器數量的增長而減少。這說明并行分布式環境在相同數據集相同任務情況下的執行效率優于單機串行環境。

② 數據量與任務執行時間的關系

在單機串行環境和三臺機器組成的分布式并行集群環境下運行相同任務。使用4組數據，每組數據包括2個基因數據文件。4組數據中每個基因數據文件分別包含4 946 865 條數據、12 367 134 條數據、24 734 269條數據和74 202 843條數據。結果如圖12 所示?？芍谝唤M數據外，分布式并行集群環境下其余組執行任務消耗的時間均小于單機串行環境。因為分布式并行集群啟動時需要初始化，所以執行第一組數據的任務時，分布式并行集群環境消耗的時間大于單機串行環境消耗的時間。由圖12可說明，分布式并行集群環境與單機串行環境相比更適合大數據量的情況。數據量越多，集群環境所表現的性能越出色。

③ CPU內核使用率

CPU執行計算。其內部有多個核心，每個核心可獨立執行任務。對于一次計算任務，使用的核心越多則使用的計算資源越多，計算速度越快。對于此評價指標，本文使用單機串行環境和三臺機器組成的分布式并行集群環境運行相同任務。使用4組數據，每組數據包括2個基因數據文件。4組數據中每個基因數據文件分別包含4 946 865 條數據、12 367 134 條數據、24 734 269條數據和74 202 843條數據。結果如圖13所示。結果表明，單機串行執行任務總是單線程的，既只使用單個CPU內核。而分布式并行集群執行任務則會使用所有CPU內核。由此表明，分布式并行環境相比于單機串行環境能更有效地使用計算資源，更快的執行任務。

④ 內存利用率

原始基因數據存儲在分布式文件系統HDFS中。當分布式并行化Cochran-Armitage算法運行時加載的數據，以及對數據處理產生的中間結果都會放在內存中。內存作為計算機資源的一種，也應該盡量降低利用率，以使內存為更多的程序使用。如圖14是在單機串行環境與分布式并行集群環境中對4組數據的內存利用率對比。每組數據包括2個基因數據文件。4組數據中每個基因數據文件分別包含4 946 865條數據、12 367 134條數據、24 734 269條數據和74 202 843條數據。由此表明，單機串行環境對內存的消耗隨著數據量的增多而顯著增加。分布式并行環境對內存的消耗隨著數據量的增加并沒有太明顯的變化。分布式并行環境對內存的消耗明顯的小于單機串行環境。

5 結論

本文設計了分布式并行Cochran-Armitage趨勢檢驗算法的Map階段的計算過程與Reduce階段的計算過程，并將任務提交至大數據計算框架Spark，執行Map和Reduce計算過程。本文使用了四組數據集，每組數據包括2個基因數據文件。4組數據中每個基因數據文件分別包含4 946 865條數據、12 367 134條數據、24 734 269條數據和74 202 843條數據。4組數據均使用串行Cocharan-Armitage 算法和分布式并行Cochran-Armitage算法進行了對比。通過上述的4個評價指標可得出結論，分布式并行的Cochran-Armitage 算法明顯優于傳統單機串行的Cochran-Armitage算法。分布式并行算法相比于串行算法可在很大程度上減少計算時間以及更合理利用和分配計算資源。

【通聯編輯：王力】