GSP290

總覽

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

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

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

學習內容

本實驗室將說明如何建構及執行 Dataflow 管道 (Python)，用來完成下列作業：

• 將 Cloud Storage 中的資料擷取至 BigQuery。
• 在 BigQuery 轉換及充實資料。
• 在 BigQuery join 資料，然後將結果寫入新的資料表。

設定和需求

瞭解以下事項後，再點選「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 目前已啟用，仍須按照下方步驟停用 API，然後再次啟用，確保成功重新啟動 API。

gcloud services disable dataflow.googleapis.com --project {{{project_0.project_id|Project ID}}} --force gcloud services enable dataflow.googleapis.com --project {{{project_0.project_id|Project ID}}}

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

點選「Check my progress」，確認工作已完成。 停用並重新啟用 Dataflow API。

工作 2：下載範例程式碼

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

• 在 Cloud Shell 執行下列指令，從 Google Cloud Professional Services GitHub 下載 Dataflow Python 範例程式碼：

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

工作 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_0.project_id | BUCKET_NAME }}} --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_0.project_id | BUCKET_NAME }}}/data_files/ gcloud storage cp gs://spls/gsp290/data_files/head_usa_names.csv gs://{{{ project_0.project_id | BUCKET_NAME }}}/data_files/

點選「Check my progress」，確認工作已完成。 將檔案複製到 bucket。

工作 4：建立 BigQuery 資料集

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

• 在 Cloud Shell 建立資料集 lake：

bq mk lake

點選「Check my progress」，確認工作已完成。 建立 BigQuery 資料集 lake。

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

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

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

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

在 Cloud Shell IDE 啟用 Gemini Code Assist

您可以在 Cloud Shell 等整合式開發環境 (IDE) 中使用 Gemini Code Assist，取得程式碼相關指引或解決程式碼問題。Gemini Code Assist 必須先啟用，才能開始使用。

1. 在 Cloud Shell 執行下列指令，啟用 Gemini for Google Cloud API：

gcloud services enable cloudaicompanion.googleapis.com

2. 點按 Cloud Shell 工具列中的「開啟編輯器」。

注意：如要開啟 Cloud Shell 編輯器，請點選 Cloud Shell 工具列中的「開啟編輯器」。如要在 Cloud Shell 與程式碼編輯器之間切換，請視需要點選「開啟編輯器」或「開啟終端機」。

3. 點選畫面底部狀態列中的「Cloud Code - No Project」。

4. 依指示授權外掛程式。如果系統未自動選取專案，請點按「選取 Google Cloud 專案」，然後選擇 。

5. 確認狀態列的 Cloud Code 狀態訊息中已顯示 Google Cloud 專案 ()。

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

在本節中，您會提示 Gemini Code Assist 分享資料擷取管道的詳細資訊，以便向新團隊成員提供簡介。

1. 在 Cloud Shell 編輯器的檔案總管中，依序前往「dataflow_python_examples」>「dataflow_python_examples」>「data_ingestion.py」。

2. 開啟 data_ingestion.py 檔案。這項動作會啟用 Gemini Code Assist，編輯器右上角出現 圖示時，就代表該工具已啟用。

3. 點選「Gemini Code Assist: Smart Actions」 圖示，然後選取「Explain this」。

4. Gemini Code Assist 會開啟對話窗格，並預先填入提示詞 Explain this。在 Code Assist 對話的內嵌文字方塊中，將預先填入的提示詞改成下列內容，然後點選「Send」：

You are an expert Data Engineer at Cymbal AI. A new team member is unfamiliar with this pipeline code. Explain the purpose and functionality of the data ingestion pipeline defined in the data_ingestion.py. Your explanation should include: 1. A high-level summary of what the script does. 2. A breakdown of the key components, such as the DataIngestion class and the run function. 3. An explanation of how the script uses the Apache Beam pipeline to read, process, and write data. 4. The role of command-line arguments and how they are used. 5. A description of the input data format and the output BigQuery table schema. For the suggested improvements, don't update this file.

這段程式碼會將 Cloud Storage 的資料檔案填入 BigQuery 資料表。Gemini Code Assist 對話視窗中會顯示 data_ingestion.py 檔案內程式碼的詳細說明。

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

設定 Dataflow 工作專用的 Docker 容器

在本節中，您將返回 Cloud Shell 工作階段，設定必要的 Python 程式庫。

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

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

cd ~ docker run -it -e PROJECT={{{ project_0.project_id | PROJECT_ID }}} -v $(pwd)/dataflow-python-examples:/dataflow python:3.8 /bin/bash

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

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

pip install apache-beam[gcp]==2.59.0

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

cd dataflow/

在雲端執行資料擷取管道

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

python dataflow_python_examples/data_ingestion.py \ --project={{{ project_0.project_id | PROJECT_ID }}} \ --region={{{ project_0.default_region | REGION }}} \ --runner=DataflowRunner \ --machine_type=e2-standard-2 \ --staging_location=gs://{{{ project_0.project_id | BUCKET_NAME }}}/test \ --temp_location gs://{{{ project_0.project_id | BUCKET_NAME }}}/test \ --input gs://{{{ project_0.project_id | BUCKET_NAME }}}/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 程式碼

在本節中，您會提示 Gemini Code Assist 分享額外的資料轉換管道資訊，進一步協助新團隊成員。

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

2. 在 Cloud Shell 編輯器中，前往同一目錄下的 data_transformation.py 檔案。如前所述，請注意編輯器右上角的 圖示。

3. 點選「Gemini Code Assist: Smart Actions」 圖示，然後選取「Explain this」。

4. Gemini Code Assist 會開啟對話窗格，並預先填入提示詞 Explain this。在 Code Assist 對話的內嵌文字方塊中，將預先填入的提示詞改成下列內容，然後點選「Send」：

You are an expert Data Engineer at Cymbal AI. A new team member is unfamiliar with this pipeline code. Explain the purpose and functionality of the data transformation pipeline defined in the data_transformation.py. Your explanation should include: 1. A high-level summary of what the script does, noting its differences from a simple ingestion pipeline. 2. A breakdown of the key components, specifically the DataTransformation class and the run function. 3. A detailed explanation of how the script uses the Apache Beam pipeline to read from a file, transform the data, and write it to a BigQuery table. 4. Describe how the script handles the BigQuery schema by reading it from a JSON file. 5. Explain the data transformation logic within the parse_method, particularly how it converts the year to a DATE type. 6. The role of command-line arguments and how they are used. For the suggested improvements, don't update this file.

Gemini Code Assist 對話視窗中會顯示 data_transformation.py 檔案內程式碼的說明。

在雲端執行資料轉換管道

1. 在 Cloud Shell 終端機中輸入下列指令，執行資料轉換管道：

python dataflow_python_examples/data_transformation.py \ --project={{{ project_0.project_id | PROJECT_ID }}} \ --region={{{ project_0.default_region | REGION }}} \ --runner=DataflowRunner \ --machine_type=e2-standard-2 \ --staging_location=gs://{{{ project_0.project_id | BUCKET_NAME }}}/test \ --temp_location gs://{{{ project_0.project_id | BUCKET_NAME }}}/test \ --input gs://{{{ project_0.project_id | BUCKET_NAME }}}/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 程式碼

在本節中，您會運用 Gemini Code Assist 的 AI 輔助功能，查看及編輯 data enrichment 管道的 Python 程式碼。

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

2. 在 Cloud Shell 編輯器中，前往同一目錄下的 data_enrichment.py。如前所述，請注意編輯器右上角的 圖示。

3. 點選工具列上的「Gemini Code Assist: Smart Actions」 圖示。

4. 將下列提示詞貼到透過工具列開啟的 Gemini Code Assist 內嵌文字欄位，更新第 83 行的程式碼。

In the data_enrichment.py file, update line 83 by replacing x.decode('utf8') with x.

5. 按下 Enter 鍵，提示 Gemini Code Assist 據此修改程式碼。

6. 在「Gemini Diff」檢視畫面出現提示時，點選「接受」。

data_enrichment.py 檔案中更新後的第 83 行程式碼，現在看起來會像這樣：

values = [x for x in csv_row]

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

執行 data enrichment 管道

1. 在 Cloud Shell 終端機中輸入下列指令，執行 data enrichment 管道：

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

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

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

3. 「工作狀態」畫面顯示 Dataflow 工作狀態為「Succeeded」後，請前往控制台，依序點選「導覽選單」圖示 >「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 資料表。

注意：如要進一步瞭解資料湖泊與資料市集 join 管道，您可以再請 Gemini Code Assist 解釋程式碼，做法和您在工作 5 及工作 6 中使用其 AI 輔助功能類似。

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_0.project_id | PROJECT_ID }}} \ --runner=DataflowRunner \ --machine_type=e2-standard-2 \ --staging_location=gs://{{{ project_0.project_id | BUCKET_NAME }}}/test \ --temp_location gs://{{{ project_0.project_id | BUCKET_NAME }}}/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 管道

隨堂測驗

請回答下列選擇題，確實掌握這個實驗室介紹的概念。按照您的理解盡力回答即可。

恭喜！

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

後續步驟/瞭解詳情

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

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

Google Cloud 教育訓練與認證

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

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

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

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