GSP290

總覽

Datafow 是 Google Cloud 推出的服務，可大規模整合串流與批次資料處理工作。這項解決方案以 Apache Beam 專案為基礎，該服務屬於開放原始碼模型，可定義批次與串流資料平行處理管道。只要使用任一開放原始碼 Apache Beam SDK，即可自行建構程式來定義管道，並運用 Dataflow 執行。

在本實驗室中，您將透過 Apache Beam SDK for Python 在 Dataflow 建構及執行管道，將 Cloud Storage 的資料擷取至 BigQuery，接著使用 BigQuery 轉換及充實資料。

注意：進行實驗室活動時，務必按照說明開啟 Python 檔案，並詳閱當中的註解，以瞭解程式碼的作用。

學習內容

本實驗室的內容包括：

• 建構並執行 Dataflow 管道 (Python)，將 Cloud Storage 的資料擷取至 BigQuery。
• 建構並執行 Dataflow 管道 (Python)，在 BigQuery 轉換及充實資料。
• 建構並執行 Dataflow 管道 (Python)，在 BigQuery 彙整資料，然後將結果寫入新的資料表。

設定和需求

瞭解以下事項後，再點選「Start Lab」按鈕

請詳閱以下操作說明。實驗室活動會計時，且中途無法暫停。點選「Start Lab」後就會開始計時，顯示可使用 Google Cloud 資源的時間。

您將在真正的雲端環境完成實作實驗室活動，而不是模擬或示範環境。為此，我們會提供新的暫時憑證，供您在實驗室活動期間登入及存取 Google Cloud。

為了順利完成這個實驗室，請先確認：

• 可以使用標準的網際網路瀏覽器 (Chrome 瀏覽器為佳)。

注意事項：請使用無痕模式 (建議選項) 或私密瀏覽視窗執行此實驗室，這可以防止個人帳戶和學員帳戶之間的衝突，避免個人帳戶產生額外費用。

• 是時候完成實驗室活動了！別忘了，活動一旦開始將無法暫停。

注意事項：務必使用實驗室專用的學員帳戶。如果使用其他 Google Cloud 帳戶，可能會產生額外費用。

如何開始研究室及登入 Google Cloud 控制台

1. 點選「Start Lab」按鈕。如果實驗室會產生費用，畫面上會出現選擇付款方式的對話方塊。左側的「Lab Details」窗格會顯示下列項目：

   • 「Open Google Cloud console」按鈕
   • 剩餘時間
   • 必須在這個研究室中使用的臨時憑證
   • 完成這個實驗室所需的其他資訊 (如有)

2. 點選「Open Google Cloud console」；如果使用 Chrome 瀏覽器，也能按一下滑鼠右鍵，選取「在無痕視窗中開啟連結」。

   接著，實驗室會啟動相關資源，並開啟另一個分頁，顯示「登入」頁面。

   提示：您可以在不同的視窗中並排開啟分頁。

   注意：如果頁面中顯示「選擇帳戶」對話方塊，請點選「使用其他帳戶」。

3. 如有必要，請將下方的 Username 貼到「登入」對話方塊。

   {{{user_0.username | "Username"}}}

   您也可以在「Lab Details」窗格找到 Username。

4. 點選「下一步」。

5. 複製下方的 Password，並貼到「歡迎使用」對話方塊。

   {{{user_0.password | "Password"}}}

   您也可以在「Lab Details」窗格找到 Password。

6. 點選「下一步」。

   重要事項：請務必使用實驗室提供的憑證，而非自己的 Google Cloud 帳戶憑證。 注意：如果使用自己的 Google Cloud 帳戶來進行這個實驗室，可能會產生額外費用。

7. 按過後續的所有頁面：

   • 接受條款及細則。
   • 由於這是臨時帳戶，請勿新增救援選項或雙重驗證機制。
   • 請勿申請免費試用。

Google Cloud 控制台稍後會在這個分頁開啟。

注意：如要使用 Google Cloud 產品和服務，請點選「導覽選單」，或在「搜尋」欄位輸入服務或產品名稱。

啟動 Cloud Shell

Cloud Shell 是搭載多項開發工具的虛擬機器，提供永久的 5 GB 主目錄，而且在 Google Cloud 中運作。Cloud Shell 提供指令列存取權，方便您使用 Google Cloud 資源。

1. 點按 Google Cloud 控制台頂端的「啟用 Cloud Shell」圖示 。

2. 系統顯示視窗時，請按照下列步驟操作：

   • 繼續操作 Cloud Shell 視窗。
   • 授權 Cloud Shell 使用您的憑證發出 Google Cloud API 呼叫。

連線建立完成即代表已通過驗證，而且專案已設為您的 Project_ID：。輸出內容中有一行文字，宣告本工作階段的 Project_ID：

Your Cloud Platform project in this session is set to {{{project_0.project_id | "PROJECT_ID"}}}

gcloud 是 Google Cloud 的指令列工具，已預先安裝於 Cloud Shell，並支援 Tab 鍵自動完成功能。

3. (選用) 您可以執行下列指令來列出使用中的帳戶：

gcloud auth list

4. 點按「授權」。

輸出內容：

ACTIVE: * ACCOUNT: {{{user_0.username | "ACCOUNT"}}} To set the active account, run: $ gcloud config set account `ACCOUNT`

5. (選用) 您可以使用下列指令來列出專案 ID：

gcloud config list project

輸出內容：

[core] project = {{{project_0.project_id | "PROJECT_ID"}}} 注意：如需 gcloud 的完整說明，請前往 Google Cloud 參閱 gcloud CLI 總覽指南。

工作 1：確定已成功啟用 Dataflow API

請重新啟動連至 Dataflow API 的連線，確保可順利使用必要的 API。

注意：即使 API 目前已啟用，仍須按照下方步驟 1 至 4 停用 API，然後再次啟用，確保成功重新啟動 API。

1. 前往 Google Cloud 控制台，在標題列的「搜尋」欄位輸入 Dataflow API，然後點選搜尋結果中的「Dataflow API」。

2. 點選「管理」。

3. 點選「停用 API」。

如果系統要求您確認操作，請點選「停用」。

4. 點選「啟用」。

API 重新啟用後，頁面上會顯示「停用」選項。

點選「Check my progress」，確認工作已完成。

停用並重新啟用 Dataflow API。

工作 2：下載範例程式碼

下載要在本實驗室使用的 Dataflow Python 範例程式碼。

1. 在 Cloud Shell 執行下列指令，從 Google Cloud 的專業服務 GitHub 下載 Dataflow Python 範例程式碼：

gcloud storage cp -r gs://spls/gsp290/dataflow-python-examples .

2. 為專案 ID 設定變數。

export PROJECT={{{ project_0.project_id }}} gcloud config set project $PROJECT

工作 3：建立 Cloud Storage bucket，並將檔案複製到 bucket

您需要在 Cloud Shell 建立 Cloud Storage bucket，然後將檔案 (即 Dataflow Python 範例程式碼) 複製到該 bucket。

建立 Cloud Storage bucket

• 繼續在 Cloud Shell 使用 make bucket 指令，為專案的 區域建立新的區域 bucket。

gcloud storage buckets create gs://$PROJECT --location={{{ project_0.default_region | REGION }}}

點選「Check my progress」，確認工作已完成。

建立 Cloud Storage bucket。

將檔案複製到 bucket

• 在 Cloud Shell 使用 gsutil 指令，將檔案複製到剛才建立的 Cloud Storage bucket：

gcloud storage cp gs://spls/gsp290/data_files/usa_names.csv gs://$PROJECT/data_files/ gcloud storage cp gs://spls/gsp290/data_files/head_usa_names.csv gs://$PROJECT/data_files/

點選「Check my progress」，確認工作已完成。

將檔案複製到 bucket。

工作 4：建立 BigQuery 資料集

在這項工作中，您將建立 BigQuery 資料集，用來存放所有載入 BigQuery 的資料表。

在 Cloud Shell 建立名為 lake 的資料集：

bq mk lake

點選「Check my progress」，確認工作已完成。

建立名為 lake 的 BigQuery 資料集。

工作 5：查看並執行資料擷取管道

在這項工作中，您將查看管道程式碼來瞭解其作用，然後設定並執行該管道。

資料擷取管道會使用 TextIO 來源和 BigQueryIO 目的地，將 Cloud Storage 的資料擷取至 BigQuery 資料表。具體來說，這個管道應能完成以下工作：

• 從 Cloud Storage 擷取檔案。
• 篩除檔案的標題列。
• 將讀取的資料行轉換為字典物件。
• 將資料列輸出至 BigQuery。

查看資料擷取管道的 Python 程式碼

請使用 Cloud Shell 程式碼編輯器查看管道程式碼。

1. 在 Cloud Shell 選單列，點選「開啟編輯器」。

2. 依序前往 dataflow_python_examples > dataflow_python_examples，然後開啟 data_ingestion.py 檔案。

3. 詳閱檔案註解，瞭解程式碼的作用。

這段程式碼會將 Cloud Storage 的資料檔案填入 BigQuery 資料表。

4. 點選「開啟終端機」，返回 Cloud Shell。

設定 Dataflow 工作專用的 Docker 容器

1. 返回 Cloud Shell 工作階段後，即可設定必要的 Python 程式庫。

這個實驗室的 Dataflow 工作需要搭配 Python3.8。為確保使用的是正確版本，您需要在 Python 3.8 Docker 容器中執行 Dataflow 程序。

2. 在 Cloud Shell 中，執行下列指令來啟動 Python 容器：

cd ~ docker run -it -e PROJECT=$PROJECT -v $(pwd)/dataflow-python-examples:/dataflow python:3.8 /bin/bash

這個指令會使用最新的 Python 3.8 穩定版提取 Docker 容器，並執行指令殼層，在容器中執行後續指令。-v 旗標會提供原始碼做為容器的 volume，讓您在 Cloud Shell 編輯器修改程式碼，同時繼續在執行中的容器內存取原始碼。

3. 容器提取完畢並開始在 Cloud Shell 執行時，請透過下列指令，在該容器中安裝 apache-beam：

pip install apache-beam[gcp]==2.59.0

4. 接著在 Cloud Shell，將執行中容器的目錄變更為連結原始碼的位置：

cd dataflow/

5. 在容器中設定專案 ID：

export PROJECT={{{ project_0.project_id }}}

在雲端執行資料擷取管道

1. 使用下列程式碼執行資料擷取管道：

python dataflow_python_examples/data_ingestion.py \ --project=$PROJECT \ --region={{{ project_0.default_region | REGION }}} \ --runner=DataflowRunner \ --machine_type=e2-standard-2 \ --staging_location=gs://$PROJECT/test \ --temp_location gs://$PROJECT/test \ --input gs://$PROJECT/data_files/head_usa_names.csv \ --save_main_session

這段程式碼會啟動所需的 worker，並在管道完成作業後將其關閉。

2. 在控制台標題列的「搜尋」欄位輸入 Dataflow，然後點選搜尋結果中的「Dataflow」。

「Dataflow」頁面開啟後，請查看工作狀態。

3. 點選工作名稱來查看進度。

「工作狀態」顯示「已完成」後，即可前往下一個步驟。這個擷取管道從啟動、完成作業到關閉，大約需要五分鐘。

4. 前往 BigQuery (依序點選「導覽選單」>「BigQuery」) 檢查資料是否確實填入。

5. 點選專案名稱，查看 lake 資料集底下的 usa_names 資料表。

6. 點選資料表，然後前往「預覽」分頁查看 usa_names 資料樣本。

注意：如果找不到 usa_names 資料表，請重新整理頁面，或使用傳統版 BigQuery 使用者介面查看。

點選「Check my progress」，確認工作已完成。

建構資料擷取管道。

工作 6：查看並執行資料轉換管道

在這項工作中，您將查看資料轉換管道來瞭解其作用，接著執行該管道以處理 Cloud Storage 檔案，最後將結果輸出至 BigQuery。

資料轉換管道也會使用 TextIO 來源和 BigQueryIO 目的地，將 Cloud Storage 的資料擷取至 BigQuery 資料表，但還會進行額外的資料轉換工作。具體來說，這個管道應能完成以下工作：

• 從 Cloud Storage 擷取檔案。
• 將讀取的資料行轉換為字典物件。
• 將包含年份的資料轉換成 BigQuery 可理解的日期格式。
• 將資料列輸出至 BigQuery。

查看資料轉換管道的 Python 程式碼

• 在程式碼編輯器，開啟 data_transformation.py 檔案。

詳閱檔案註解，瞭解程式碼的作用。

在雲端執行資料轉換管道

1. 透過下列程式碼執行資料轉換管道：

python dataflow_python_examples/data_transformation.py \ --project=$PROJECT \ --region={{{ project_0.default_region | REGION }}} \ --runner=DataflowRunner \ --machine_type=e2-standard-2 \ --staging_location=gs://$PROJECT/test \ --temp_location gs://$PROJECT/test \ --input gs://$PROJECT/data_files/head_usa_names.csv \ --save_main_session

2. 前往 Google Cloud 控制台，在標題列的「搜尋」欄位輸入 Dataflow，然後點選搜尋結果中的「Dataflow」。

3. 點選工作名稱，查看這項工作的狀態。

這個 Dataflow 管道從啟動、完成作業到關閉，大約需要五分鐘。

4. 「工作狀態」畫面顯示 Dataflow 工作狀態為「已完成」後，請前往「BigQuery」檢查資料是否確實填入。

您應該會在 lake 資料集底下看到 usa_names_transformed 資料表。

5. 點選資料表，然後前往「預覽」分頁，查看 usa_names_transformed 資料樣本。

注意：如果找不到 usa_names_transformed 資料表，請重新整理頁面，或使用傳統版 BigQuery 使用者介面查看。

點選「Check my progress」，確認工作已完成。

建構資料轉換管道。

工作 7：查看並執行 data enrichment 管道

現在您將建構 data enrichment 管道來完成下列作業：

• 從 Cloud Storage 擷取檔案。
• 篩除檔案的標題列。
• 將讀取的資料行轉換為字典物件。
• 將資料列輸出至 BigQuery。

查看並編輯 data enrichment 管道的 Python 程式碼

1. 在程式碼編輯器，開啟 data_enrichment.py 檔案。

2. 詳閱註解，瞭解程式碼的作用。這段程式碼會將資料填入 BigQuery。

目前第 83 行程式碼看起來如下所示：

values = [x.decode('utf8') for x in csv_row]

3. 請將程式碼編輯成跟下方一樣：

values = [x for x in csv_row]

4. 修改完這行程式碼後，記得在程式碼編輯器選取「檔案」選項，並點選「儲存」，儲存更新後的檔案。

執行 data enrichment 管道

1. 使用下列程式碼執行 data enrichment 管道：

python dataflow_python_examples/data_enrichment.py \ --project=$PROJECT \ --region={{{ project_0.default_region | REGION }}} \ --runner=DataflowRunner \ --machine_type=e2-standard-2 \ --staging_location=gs://$PROJECT/test \ --temp_location gs://$PROJECT/test \ --input gs://$PROJECT/data_files/head_usa_names.csv \ --save_main_session

2. 在「Dataflow」頁面，點選工作來查看狀態。

這個 Dataflow 管道從啟動、完成作業到關閉，大約需要五分鐘。

3. 「工作狀態」畫面顯示 Dataflow 工作狀態為「已完成」後，請前往控制台，依序點選「導覽選單」圖示 >「BigQuery」，檢查資料是否確實填入。

您應該會在 lake 資料集底下看到 usa_names_enriched 資料表。

4. 點選資料表，然後前往「預覽」分頁查看 usa_names_enriched 資料樣本。

注意：如果找不到 usa_names_enriched 資料表，請重新整理頁面，或使用傳統版 BigQuery 使用者介面查看。

點選「Check my progress」，確認工作已完成。

建構 data enrichment 管道。

工作 8：查看並執行資料湖泊與資料市集 join 管道

現在您將建構並使用 Dataflow 管道，從兩個 BigQuery 資料來源讀取資料，然後 join 資料來源。具體來說，這個管道應能完成以下工作：

• 從兩個 BigQuery 來源擷取檔案。
• 彙整兩個資料來源。
• 篩除檔案的標題列。
• 將讀取的資料行轉換為字典物件。
• 將資料列輸出至 BigQuery。

透過資料擷取管道執行資料 join 作業，然後將結果資料表寫入 BigQuery

首先請查看 data_lake_to_mart.py 程式碼來瞭解其作用，接著在雲端執行這個管道。

1. 在程式碼編輯器中，開啟 data_lake_to_mart.py 檔案。

請詳閱檔案註解，瞭解程式碼的作用。這段程式碼會 join 兩份資料表，並將結果寫入新的 BigQuery 資料表。

2. 使用下列程式碼區塊執行管道：

python dataflow_python_examples/data_lake_to_mart.py \ --worker_disk_type="compute.googleapis.com/projects//zones//diskTypes/pd-ssd" \ --max_num_workers=4 \ --project=$PROJECT \ --runner=DataflowRunner \ --machine_type=e2-standard-2 \ --staging_location=gs://$PROJECT/test \ --temp_location gs://$PROJECT/test \ --save_main_session \ --region={{{ project_0.default_region | REGION }}}

3. 前往 Google Cloud 控制台，在標題列的「搜尋」欄位輸入 Dataflow，然後點選搜尋結果中的「Dataflow」。

4. 點選這項新工作來查看狀態。

這個 Dataflow 管道從啟動、完成作業到關閉，大約需要五分鐘。

5. 「工作狀態」畫面顯示 Dataflow 工作狀態為「已完成」後，請依序點選「導覽選單」圖示 >「BigQuery」，檢查資料是否確實填入。

您應該會在 lake 資料集底下看到 orders_denormalized_sideinput 資料表。

6. 點選資料表，然後前往「預覽」部分查看 orders_denormalized_sideinput 資料樣本。

注意：如果找不到 orders_denormalized_sideinput 資料表，請重新整理頁面，或使用傳統版 BigQuery 使用者介面查看。

點選「Check my progress」，確認工作已完成。

建構 Dataflow 資料湖泊與資料市集 join 管道。

隨堂測驗

您可以透過下列選擇題加強自己的理解，確實掌握這個實驗室介紹的概念。按照您的理解盡力回答即可。

恭喜！

您已順利使用 Dataflow 執行 Python 程式碼，將 Cloud Storage 的資料擷取至 BigQuery，並在 BigQuery 轉換及充實資料。

後續步驟/瞭解詳情

想瞭解更多資訊嗎？歡迎瀏覽以下官方說明文件：

• Dataflow
• BigQuery
• 查看 Apache Beam 程式設計指南，學習更多進階概念。
• 參考下列實驗室：
  • 使用 BQML 中的分類模型預測訪客購買行為
  • Cloud Composer：複製不同位置的 BigQuery 資料表

Google Cloud 教育訓練與認證

協助您瞭解如何充分運用 Google Cloud 的技術。我們的課程會介紹專業技能和最佳做法，讓您可以快速掌握要領並持續進修。我們提供從基本到進階等級的訓練課程，並有隨選、線上和虛擬課程等選項，方便您抽空參加。認證可協助您驗證及證明自己在 Google Cloud 技術方面的技能和專業知識。

使用手冊上次更新日期：2025 年 4 月 1 日

實驗室上次測試日期：2025 年 4 月 1 日

Copyright 2025 Google LLC 保留所有權利。Google 和 Google 標誌是 Google LLC 的商標，其他公司和產品名稱則有可能是其關聯公司的商標。