GCP的資料流水線平台DataFlow

本文將介紹Apache Beam 和 Dataflow 模組。 Dataflow 是 Apache Beam 的補充技術，它們都可以幫助您我們建立和運行預處理和特徵工程。

Cloud Dataflow

考量特徵預處理甚至任何資料轉換的方法之一是從流水線的角度來思考。這裡說的流水線是指的是將資料從一種格式轉換為另一種格式的一系列步驟。

假設我們在 BigQuery有一些資料。可以使用 BigQuery 作為流水線的輸入，執行一系列步驟來轉換資料，並可能在轉換過程中引入一些新功特徵。最後，可以將結果儲存到輸出，例如 Cloud Storage。而Cloud Dataflow 是一個讓我們運行這些類型的資料處理流水線的平台。 Dataflow 可以運行用 Python 和 Java 程式語言編寫的流水線。 Dataflow 將自己定位為資料轉換平台，因為它是 Google 提供的無伺服器、完全託管的產品，可大規模執行資料處理流水線。

為 Dataflow 編寫代碼的方式是使用 Apache Beam 的開源函式庫。因此，要實作資料處理流水線，我們可以使用 Apache Beam API 編寫代碼，然後將代碼部署到 Cloud Dataflow。 Apache Beam 易於使用的一點是，為 Beam 編寫的代碼與技術人員對資料處理流水線的看法類似。

如上圖中央的流水線。此範例 Python 代碼分析檔案中文字行中的單字數。因此，作為流水線的輸入，可能是從 Cloud Storage 讀取文字檔案。然後轉換資料，找出每行文字的單字數。這種轉換可以透過 Dataflow 自動擴展以平行運作。接下來在流水線中，可以使用分組和其他聚合操作按單字數對行進行分組。

我們也可以過濾掉一些值，例如忽略少於 10 個單字的行。一旦完成所有轉換、分組和過濾操作，流水線就會將結果寫入Cloud Storage。注意，此實作將流水線定義與流水線執行分開。在呼叫 p.run 方法之前看到的所有步驟只是定義流水線應該做什麼。只有當呼叫 run 方法時，流水線才會真正執行。

Apache Beam 的功能之一是它使用相同的流水線代碼支援批次和串流資料處理。事實上，這個函式庫的名字 Beam 來自於 batch 和 stream 的縮寫。無論資料是來自大量資料來源（如 Cloud Storage），還是來自串流資料來源（如 PubSub），我們都可以重複使用相同的流水線邏輯。

Dataflow的術語與概念

上面我們已經說明了運行資料流的資料處理流水線。在上途中，我們可以看到流水線的圖形。流水線必須有一個來源，即管流水線從那裡獲取輸入資料。該流水線有一系列的步驟。 Beam 中的每個步驟稱為一次變換(transform)。每個轉換都在一個稱為 PCollection 的結構上進行。現在，我們只需記住每個轉換都以一個 PCollection 作為輸入，並將結果輸出到另一個 PCollection。流水線中最後一次轉換的結果很重要。它進入同步狀態，這是流水線的輸出。

要運作流水線，需要一個稱為runner的東西。Runner取得流水線代碼並執行。Runner是特定於平台的，這意味著有一個"資料流runner"用於在Dataflow上執行流水線。如果想使用 Apache Spark 來運作流水線，還要有另一個Runner來支援。還有一個direct runner，它將在我們的本機電腦上執行流水線。如果我們想要，甚至可以為自己的分散式運算平台實作自己的custom runner。

那麼如何實現這些流水線？如果我們看一下上圖的代碼，我們會注意到主方法中的流水線操作是beam。流水線會去建立流水線實例。一旦建立，每個轉換都將作為流水線應用方法的參數來實現。

在 Python 版本的 Apache Beam 函式庫中，pipe operator被重載以呼叫 apply method。這就是為什麼會有這種將pipe operator相互疊加在一起的奇怪語法。

諸如 Read、CountWords 和 Write 之類的字串只是我們可以為流水線中的每個轉換人類可以理解的名稱。正如我們上面提到的，沒有任何pipeline operators真正運作該流水線。當我們需要流水線處理一些資料時，我們需要呼叫pipeline instance上的 run 方法來執行它。

關於 PCollections，需要了解的一件重要事情是，與許多資料結構不同，PCollection 不會儲存其所有的資料記憶體。因此，PCollection 就像一個資料結構，帶有指向dataflow cluster儲存資料的位置指標。