如何用Google Cloud 建立簡易數(shù)據(jù)處理流程

針對不同的原始數(shù)據(jù)類型-批數(shù)據(jù)（Batch Data)或流數(shù)據(jù)（Streaming Data），處理數(shù)據(jù)的流程也會略有不同。本文通過一個簡單的例子，來介紹如何借助Google Cloud建立批數(shù)據(jù)處理流程。

01丨提出需求

假設(shè)我們想要建立一個數(shù)據(jù)集，其包含大陸兩大證券交易所（上海及深圳證券交易所）每日成交量排名前20的股票信息。

原始信息可以通過兩大證券交易所網(wǎng)站獲得：

https://www.sse.com.cn/market/stockdata/activity/main/

https://www.szse.cn/market/stock/active/actv/index.html

02丨分析需求

在實現(xiàn)代碼前，我們需要先對需求進行分析，建立模型，并判斷需要實現(xiàn)的功能。

建模

一個典型的ETL（Extract,Transform,Load）數(shù)據(jù)流程大都由以下幾部分組成：

·原始數(shù)據(jù)存儲（download）從數(shù)據(jù)源（Vendor Source）下載數(shù)據(jù)，并備份到自有存儲。自有存儲的好處是便于日后重新處理或查找，而不用擔心數(shù)據(jù)源的消失。

·原始數(shù)據(jù)變形（reshape）將不同數(shù)據(jù)格式統(tǒng)一，重新生成原始數(shù)據(jù)。不同數(shù)據(jù)提供商的數(shù)據(jù)保存格式往往不同（如csv，excel，xml等），將不同的數(shù)據(jù)格式進行統(tǒng)一，能方便后續(xù)的數(shù)據(jù)處理。

·數(shù)據(jù)歸一（normalize）對變形后的數(shù)據(jù)進行重新建模，在理想的數(shù)據(jù)模型（schema）下重新生成數(shù)據(jù)。不同數(shù)據(jù)源的數(shù)據(jù)，其數(shù)據(jù)模型往往不盡相同，同時與我們最終想要得到的數(shù)據(jù)集模型也未必相同，所以我們需要對數(shù)據(jù)進行歸一處理。

·數(shù)據(jù)整合（merge）對不同來源的相同數(shù)據(jù)進行整合，生成最終數(shù)據(jù)集。對同一對象，不同的數(shù)據(jù)源可能會提供不同的數(shù)據(jù)。例如根據(jù)計算方法不同，不同數(shù)據(jù)提供商針對同一支證券會提供不同的的收盤價。我們需要將這些數(shù)據(jù)整合來生成最終的數(shù)據(jù)集。

分析

在本例中，數(shù)據(jù)源（Vendor Source）共有兩個，上海證券交易所和深圳證券交易所。

上海證券交易所提供的數(shù)據(jù)存儲在html文件中，除了成交量的排名外還有成交金額等其他指標排名。而深圳證券交易所提供的數(shù)據(jù)存儲在xlsx文件中，并僅有成交量的排名。

上海證券交易所提供的數(shù)據(jù)中除了股票代碼，股票簡稱，累計成交量外，還有價格信息（開盤，收盤，均價）及幾個比率信息（振幅，換手率）等。而深圳證券交易所提供的數(shù)據(jù)除了股票代碼，股票簡稱，累計成交量外，也提供幾個價格信息。同時兩組數(shù)據(jù)的單位并不相同（上海證券交易所為萬股、萬元，深圳證券交易所為億股、億元）。

兩組數(shù)據(jù)互斥（mutually exclusive），所以假設(shè)如果沒有人為改寫（manual override）的需求，數(shù)據(jù)整合并不需要做什么。

根據(jù)上面的分析，我們確定數(shù)據(jù)流程中需要完成以下幾項任務(wù)：

·下載html及xlsx文件。

·轉(zhuǎn)換html及xslx文件至同一格式。

·對兩組數(shù)據(jù)進行重新建模，統(tǒng)一數(shù)據(jù)模型及單位。

·簡單數(shù)據(jù)整合并輸出數(shù)據(jù)集。

流程圖

根據(jù)上面的分析，我們可以建立以下流程圖：

03丨實現(xiàn)需求

技術(shù)棧

在本例中，數(shù)據(jù)流程的實現(xiàn)需要用到Google Cloud中的以下幾個產(chǎn)品：

·Cloud Storage用來存儲原始數(shù)據(jù)文件。

·Cloud Function用來下載數(shù)據(jù)文件，觸發(fā)文件存儲后的各項變形。

·Cloud Scheduler用來定時執(zhí)行Cloud Function。

·Dataflow用來執(zhí)行數(shù)據(jù)變形，歸一以及整合。Big Query用來存儲變形后，歸一后，以及整合后的數(shù)據(jù)。

原始數(shù)據(jù)存儲

上海證券交易所主板成交量前20的數(shù)據(jù)并沒有提供文件下載鏈接，所以我們只能存儲整個活躍股排名前20的html頁面。

數(shù)據(jù)下載及存儲可以通過Cloud Function，Cloud Scheduler和Cloud Storage來實現(xiàn)

Cloud Function偽代碼如下:

通過Cloud Scheduler定時調(diào)用上面的Cloud Function，下載每日的數(shù)據(jù)信息，并將最終的html文件存儲在Cloud Storage中。

深圳證券交易所主板成交量前20的數(shù)據(jù)提供xlsx格式下載。同上，我們依舊可以通過Cloud Function，Cloud Scheduler和Cloud Storage來實現(xiàn)

Cloud Function偽代碼如下:

原始數(shù)據(jù)變形，歸一以及整合

在原始文件下載后，我們需要對原始數(shù)據(jù)進行變形，歸一以及整合。這些變形都可以通過Dataflow來實現(xiàn)。每一個數(shù)據(jù)源，我們都需要進行變形及歸一操作，而整合只需要對所有數(shù)據(jù)進行統(tǒng)一處理便可。

針對上海證券交易所數(shù)據(jù)的變形、歸一的偽代碼如下：

針對深圳證券交易所數(shù)據(jù)的變形、歸一與上面類似。

之后，我們可以通過Cloud Function來建立觸發(fā)機制（trigger），每當原始數(shù)據(jù)文件被下載到Google Cloud Storage后，上面的Dataflow流程便會被自動調(diào)用。

因為本例中數(shù)據(jù)互斥，數(shù)據(jù)整合僅僅是簡單的合并，所以整合部分的偽代碼便略過了。

最后，整個流程中生成的各種數(shù)據(jù)會被存儲到Big Query的各個表格中：

main_board_trading_volume_top_20中存儲的數(shù)據(jù)便是我們需要的數(shù)據(jù)集。

04丨寫在最后

如果要將上面所建立的數(shù)據(jù)流程應用到正式的生產(chǎn)環(huán)境（Production Environment）中，其實還有大量的問題需要解決，以下僅列出幾個例子：

·技術(shù)棧的選取隨著各種技術(shù)的不斷發(fā)展，我們往往有不同的選擇來實現(xiàn)同一功能。如存儲數(shù)據(jù)我們可以用Big Query，也可以用RDBMS，NoSQL等。選取的技術(shù)棧是否合適，是我們需要仔細考慮的問題。

·代碼復用（Code Reusability）處理簡單數(shù)據(jù)的數(shù)據(jù)流程大都大同小異。為了能減少后期維護成本，以及提高代碼復用，這些數(shù)據(jù)流程往往是通過一個共同的數(shù)據(jù)流程框架來實現(xiàn)，而不是成百上千個獨立的小程序，所以上面示例中存在的大量重復代碼，在正式生產(chǎn)環(huán)境中并不常見。

·證券標識（Security ID）統(tǒng)一一個量化交易系統(tǒng)的搭建，往往需要數(shù)十乃至上百個數(shù)據(jù)集的協(xié)作。同一支證券，在不同數(shù)據(jù)集中往往擁有不同的標示（如Ticker，SEDOL，ISIN，CUSIP等）。為了能將新的數(shù)據(jù)集導入現(xiàn)有平臺，我們需要解決證券標識的統(tǒng)一問題。

·流程的可靠性（Pipeline Reliability）及錯誤處理(Error Handling)一個流程的可靠性會極大地影響后期的維護成本。上面的例子中，一個常見的改動（文件地址更新，不完整文件，數(shù)據(jù)錯誤，數(shù)據(jù)模型更改）便會使整個數(shù)據(jù)流程出錯。我們需要盡可能的完善錯誤處理機制，來盡可能的減少后期維護需求。