使用工作流程將資料從 Cloud Storage 載入至 BigQuery

Last reviewed 2021-05-12 UTC

本教學課程說明如何使用 Workflows、Cloud Run functions 和 Firestore，以無伺服器工作流程可靠地從 Cloud Storage 將原始資料 (例如事件記錄) 載入 BigQuery。數據分析平台通常會提供自動化調度管理工具，定期使用 BigQuery 工作將資料載入 BigQuery，然後轉換資料，並使用 SQL 陳述式 (包括 BigQuery 程序語言陳述式) 提供業務指標。本教學課程適用於想建構無伺服器事件導向資料處理管道的開發人員和架構師。本教學課程假設您熟悉 YAML、SQL 和 Python。

架構

下圖顯示無伺服器擷取、載入及轉換 (ELT) 管道的整體架構，該管道使用 Workflows。

在上圖中，假設零售平台會定期從各家商店收集銷售事件 (以檔案形式)，然後將檔案寫入 Cloud Storage bucket。這些事件會匯入 BigQuery 並經過處理，以提供業務指標。這個架構提供可靠的無伺服器自動化調度管理系統，可將檔案匯入 BigQuery，並分為下列兩個模組：

檔案清單：維護新增至 Firestore 集合中 Cloud Storage 值區的未處理檔案清單。這個模組會透過 Cloud Run 函式運作，並由物件完成儲存空間事件觸發，當新檔案新增至 Cloud Storage bucket 時，就會產生這個事件。檔案名稱會附加至 Firestore 中名為 new 的集合陣列。files
工作流程：執行排定的工作流程。Cloud Scheduler 會觸發工作流程，根據 YAML 語法執行一系列步驟，協調載入作業，然後呼叫 Cloud Run 函式，在 BigQuery 中轉換資料。工作流程中的步驟會呼叫 Cloud Run 函式，執行下列工作：
- 建立並啟動 BigQuery 載入工作。
- 輪詢載入工作狀態。
- 建立並啟動轉換查詢工作。
- 輪詢轉換工作狀態。

使用交易在 Firestore 中維護新檔案清單，有助於確保工作流程將檔案匯入 BigQuery 時不會遺漏任何檔案。工作流程的個別執行作業會將工作的中繼資料和狀態儲存在 Firestore 中，藉此確保作業的冪等性。

目標

建立 Firestore 資料庫。
設定 Cloud Run 函式觸發條件，追蹤新增至 Firestore 中 Cloud Storage bucket 的檔案。
部署 Cloud Run 函式，執行及監控 BigQuery 工作。
部署及執行工作流程，自動化處理程序。

費用

在本文件中，您會使用下列 Google Cloud的計費元件：

如要根據預測用量估算費用，請使用 Pricing Calculator。

初次使用 Google Cloud 的使用者可能符合免費試用期資格。

完成本文所述工作後，您可以刪除建立的資源，避免繼續計費，詳情請參閱「清除所用資源」。

事前準備

在 Google Cloud 控制台的專案選擇器頁面中，選取或建立 Google Cloud 專案。
選取或建立專案所需的角色
- 選取專案：選取專案時，不需要具備特定 IAM 角色，只要您已獲授角色，即可選取任何專案。
- 建立專案：如要建立專案，您需要具備專案建立者角色 (roles/resourcemanager.projectCreator)，其中包含 resourcemanager.projects.create 權限。瞭解如何授予角色。
注意：如果您不打算保留在這項程序中建立的資源，請建立新專案，而不要選取現有專案。完成這些步驟後，您就可以刪除專案，並移除與該專案相關聯的所有資源。

前往專案選取器
確認專案已啟用計費功能 Google Cloud 。
啟用 Cloud Build、Cloud Run 函式、Identity and Access Management、Resource Manager 和 Workflows API。
啟用 API 時所需的角色
如要啟用 API，您需要服務使用情形管理員 IAM 角色 (roles/serviceusage.serviceUsageAdmin)，其中包含 serviceusage.services.enable 權限。瞭解如何授予角色。
啟用 API
前往「Welcome」頁面，記下「專案 ID」，以供後續步驟使用。

前往歡迎頁面
在 Google Cloud 控制台中啟用 Cloud Shell。

啟用 Cloud Shell

準備環境

如要準備環境，請建立 Firestore 資料庫、從 GitHub 存放區複製程式碼範例、使用 Terraform 建立資源、編輯 Workflows YAML 檔案，以及安裝檔案產生器的必要條件。

如要建立 Firestore 資料庫，請按照下列步驟操作：
1. 前往 Google Cloud 控制台的 Firestore 頁面。
  
  前往 Firestore
2. 按一下「選取原生模式」。
3. 在「選取位置」選單中，選取要代管 Firestore 資料庫的區域。建議選擇靠近您實際位置的區域。
4. 按一下 [Create database] (建立資料庫)。

在 Cloud Shell 中，複製來源存放區：

cd $HOME && git clone https://github.com/GoogleCloudPlatform/workflows-demos
cd workflows-demos/workflows-bigquery-load

在 Cloud Shell 中，使用 Terraform 建立下列資源：
```
terraform init
terraform apply \
    -var project_id=PROJECT_ID \
    -var region=REGION \
    -var zone=ZONE \
    --auto-approve
```
更改下列內容：
- PROJECT_ID：您的 Google Cloud 專案 ID
- REGION：用於代管資源的特定 Google Cloud地理位置，例如 us-central1
- ZONE：用於代管資源的區域內位置，例如 us-central1-b
畫面會顯示類似以下的訊息： Apply complete! Resources: 7 added, 0 changed, 1 destroyed.

Terraform 可協助您以安全且可預測的方式，大規模建立、變更及升級基礎架構。專案中會建立下列資源：
- 服務帳戶：具備必要權限，可確保資源存取安全。
- 名為 serverless_elt_dataset 的 BigQuery 資料集，以及名為 word_count 的資料表，用於載入傳入的檔案。
- 名為 ${project_id}-ordersbucket 的 Cloud Storage bucket，用於暫存輸入檔案。
- 下列五個 Cloud Run 函式：
  - file_add_handler 會將新增至 Cloud Storage bucket 的檔案名稱新增至 Firestore 集合。
  - create_job 會建立新的 BigQuery 載入工作，並將 Firebase 集合中的檔案與該工作建立關聯。
  - create_query 會建立新的 BigQuery 查詢工作。
  - poll_bigquery_job 會取得 BigQuery 工作的狀態。
  - run_bigquery_job 會啟動 BigQuery 工作。

取得您在上一個步驟中部署的 create_job、create_query、poll_job 和 run_bigquery_job Cloud Run 函式網址。

gcloud functions describe create_job | grep url
gcloud functions describe poll_bigquery_job | grep url
gcloud functions describe run_bigquery_job | grep url
gcloud functions describe create_query | grep url

輸出結果會與下列內容相似：

url: https://REGION-PROJECT_ID.cloudfunctions.net/create_job
url: https://REGION-PROJECT_ID.cloudfunctions.net/poll_bigquery_job
url: https://REGION-PROJECT_ID.cloudfunctions.net/run_bigquery_job
url: https://REGION-PROJECT_ID.cloudfunctions.net/create_query

請記下這些網址，部署工作流程時會用到。

建立及部署工作流程

在 Cloud Shell 中開啟工作流程的來源檔案： workflow.yaml：

main:
  steps:
    - constants:
        assign:
          - create_job_url: CREATE_JOB_URL
          - poll_job_url: POLL_BIGQUERY_JOB_URL
          - run_job_url: RUN_BIGQUERY_JOB_URL
          - create_query_url: CREATE_QUERY_URL
          - region: BQ_REGION
          - table_name: BQ_DATASET_TABLE_NAME
        next: createJob

    - createJob:
        call: http.get
        args:
          url: ${create_job_url}
          auth:
              type: OIDC
          query:
              region: ${region}
              table_name: ${table_name}
        result: job
        next: setJobId

    - setJobId:
        assign:
          - job_id: ${job.body.job_id}
        next: jobCreateCheck

    - jobCreateCheck:
        switch:
          - condition: ${job_id == Null}
            next: noOpJob
        next: runLoadJob

    - runLoadJob:
        call: runBigQueryJob
        args:
            job_id: ${job_id}
            run_job_url: ${run_job_url}
            poll_job_url: ${poll_job_url}
        result: jobStatus
        next: loadRunCheck

    - loadRunCheck:
        switch:
          - condition: ${jobStatus == 2}
            next: createQueryJob
        next: failedLoadJob

    - createQueryJob:
        call: http.get
        args:
          url: ${create_query_url}
          query:
              qs: "select count(*) from serverless_elt_dataset.word_count"
              region: "US"
          auth:
              type: OIDC
        result: queryjob
        next: setQueryJobId

    - setQueryJobId:
        assign:
          - qid: ${queryjob.body.job_id}
        next: queryCreateCheck

    - queryCreateCheck:
        switch:
          - condition: ${qid == Null}
            next: failedQueryJob
        next: runQueryJob

    - runQueryJob:
        call: runBigQueryJob
        args:
          job_id: ${qid}
          run_job_url: ${run_job_url}
          poll_job_url: ${poll_job_url}
        result: queryJobState
        next: runQueryCheck

    - runQueryCheck:
        switch:
          - condition: ${queryJobState == 2}
            next: allDone
        next: failedQueryJob

    - noOpJob:
        return: "No files to import"
        next: end

    - allDone:
        return: "All done!"
        next: end

    - failedQueryJob:
        return: "Query job failed"
        next: end

    - failedLoadJob:
        return: "Load job failed"
        next: end


runBigQueryJob:
  params: [job_id, run_job_url, poll_job_url]
  steps:
    - startBigQueryJob:
        try:
          call: http.get
          args:
              url: ${run_job_url}
              query:
                job_id: ${job_id}
              auth:
                type: OIDC
              timeout: 600
          result: submitJobState
        retry: ${http.default_retry}
        next: validateSubmit

    - validateSubmit:
        switch:
          - condition: ${submitJobState.body.status == 1}
            next: sleepAndPollLoad
        next: returnState

    - returnState:
        return: ${submitJobState.body.status}

    - sleepAndPollLoad:
        call: sys.sleep
        args:
          seconds: 5
        next: pollJob

    - pollJob:
        try:
          call: http.get
          args:
            url: ${poll_job_url}
            query:
              job_id: ${job_id}
            auth:
              type: OIDC
            timeout: 600
          result: pollJobState
        retry:
          predicate: ${http.default_retry_predicate}
          max_retries: 10
          backoff:
            initial_delay: 1
            max_delay: 60
            multiplier: 2
        next: stateCheck

    - stateCheck:
        switch:
          - condition: ${pollJobState.body.status == 2}
            return: ${pollJobState.body.status}
          - condition: ${pollJobState.body.status == 3}
            return: ${pollJobState.body.status}
        next: sleepAndPollLoad

更改下列內容：

CREATE_JOB_URL：用於建立新工作的函式網址
POLL_BIGQUERY_JOB_URL：輪詢執行中工作狀態的函式網址
RUN_BIGQUERY_JOB_URL：啟動 BigQuery 載入工作的函式網址
CREATE_QUERY_URL：啟動 BigQuery 查詢工作的函式網址
BQ_REGION：資料儲存所在的 BigQuery 區域，例如 US
BQ_DATASET_TABLE_NAME：BigQuery 資料集資料表名稱，格式為 PROJECT_ID.serverless_elt_dataset.word_count

部署 workflow 檔案：

gcloud workflows deploy WORKFLOW_NAME \
    --location=WORKFLOW_REGION \
    --description='WORKFLOW_DESCRIPTION' \
    --service-account=workflow-runner@PROJECT_ID.iam.gserviceaccount.com \
    --source=workflow.yaml

更改下列內容：

WORKFLOW_NAME：工作流程的專屬名稱
WORKFLOW_REGION：工作流程部署所在的區域，例如 us-central1
WORKFLOW_DESCRIPTION：工作流程說明

建立 Python 3 虛擬環境，並安裝檔案產生器的必要元件：

sudo apt-get install -y python3-venv
python3 -m venv env
. env/bin/activate
cd generator
pip install -r requirements.txt

產生要匯入的檔案

gen.py Python 指令碼會以 Avro 格式產生隨機內容。結構定義與 BigQuery word_count 資料表相同。這些 Avro 檔案會複製到指定的 Cloud Storage bucket。

在 Cloud Shell 中產生檔案：

python gen.py -p PROJECT_ID \
    -o PROJECT_ID-ordersbucket \
    -n RECORDS_PER_FILE \
    -f NUM_FILES \
    -x FILE_PREFIX

更改下列內容：

RECORDS_PER_FILE：單一檔案中的記錄數
NUM_FILES：要上傳的檔案總數
FILE_PREFIX：產生檔案名稱的前置字串

在 Firestore 中查看檔案項目

檔案複製到 Cloud Storage 時，會觸發 handle_new_file Cloud Run 函式。這個函式會將檔案清單新增至 Firestore jobs 集合中 new 文件內的檔案清單陣列。

如要查看檔案清單，請前往 Google Cloud 控制台的 Firestore「資料」頁面。

前往「資料」

觸發工作流程

Workflows 會將一系列無伺服器工作串連在一起，這些工作來自Google Cloud 和 API 服務。這項工作流程中的每個步驟都會以 Cloud Run 函式形式執行，狀態則會儲存在 Firestore 中。所有對 Cloud Run functions 的呼叫都會使用工作流程的服務帳戶進行驗證。

在 Cloud Shell 中執行工作流程：

gcloud workflows execute WORKFLOW_NAME

下圖顯示工作流程中使用的步驟：

工作流程分為兩部分：主要工作流程和子工作流程。主要工作流程會處理工作建立和條件式執行作業，而子工作流程則會執行 BigQuery 工作。工作流程會執行下列作業：

create_job Cloud Run 函式會建立新的工作物件，從 Firestore 文件取得新增至 Cloud Storage 的檔案清單，並將檔案與載入工作建立關聯。如果沒有要載入的檔案，函式就不會建立新工作。
create_query Cloud Run 函式會接收需要執行的查詢，以及查詢應執行的 BigQuery 區域。函式會在 Firestore 中建立工作，並傳回工作 ID。
run_bigquery_job Cloud Run 函式會取得需要執行的工作 ID，然後呼叫 BigQuery API 提交工作。
您不必在 Cloud Run 函式中等待工作完成，可以定期輪詢工作狀態。
- poll_bigquery_job Cloud Run 函式會提供工作狀態。系統會重複呼叫這個函式，直到工作完成為止。
- 如要在呼叫 poll_bigquery_job Cloud Run 函式之間加入延遲時間，請從 Workflows 呼叫 sleep 常式。

查看工作狀態

您可以查看檔案清單和工作狀態。

前往Google Cloud 控制台的 Firestore「資料」頁面。

前往「資料」
系統會為每項工作產生專屬 ID (UUID)。如要查看 job_type 和 status，請按一下工作 ID。每項工作可能具有下列其中一種類型和狀態：
- job_type：工作流程執行的工作類型，可能值如下：
  - 0：將資料載入 BigQuery。
  - 1：在 BigQuery 中執行查詢。
- status：工作目前的狀態，可能的值如下：
  - 0：工作已建立，但尚未啟動。
  - 1：工作正在執行。
  - 2：工作順利執行完畢。
  - 3：發生錯誤，工作未順利完成。
工作物件也包含中繼資料屬性，例如 BigQuery 資料集所屬的區域、BigQuery 資料表的名稱，以及 (如果是查詢工作) 執行的查詢字串。

查看 BigQuery 中的資料

如要確認 ELT 工作是否成功，請確認資料是否顯示在資料表中。

前往 Google Cloud 控制台的「BigQuery 編輯器」頁面。

前往編輯器
按一下「serverless_elt_dataset.word_count」資料表。
點選「預覽」分頁標籤，

排定工作流程

如要依排程定期執行工作流程，可以使用 Cloud Scheduler。

清除所用資源

如要避免付費，最簡單的方法就是刪除您為本教學課程建立的專案。 Google Cloud 或者，您也可以刪除個別資源。

刪除個別資源

在 Cloud Shell 中，移除使用 Terraform 建立的所有資源：

cd $HOME/bigquery-workflows-load
terraform destroy \
-var project_id=PROJECT_ID \
-var region=REGION \
-var zone=ZONE \
--auto-approve

前往 Google Cloud 控制台的 Firestore「資料」頁面。

前往「資料」
按一下「工作」旁邊的「選單」，然後選取「刪除」。

刪除專案

前往 Google Cloud 控制台的「Manage resources」(管理資源) 頁面。
前往「Manage resources」(管理資源)
在專案清單中選取要刪除的專案，然後點選「Delete」(刪除)。
在對話方塊中輸入專案 ID，然後按一下 [Shut down] (關閉) 以刪除專案。

後續步驟

如要進一步瞭解 BigQuery，請參閱 BigQuery 說明文件。
瞭解如何建構無伺服器自訂機器學習管道。
如要查看更多參考架構、圖表和最佳做法，請瀏覽 Cloud Architecture Center。

使用工作流程將資料從 Cloud Storage 載入至 BigQuery 透過集合功能整理內容 你可以依據偏好儲存及分類內容。

架構

目標

費用

事前準備

準備環境

建立及部署工作流程

產生要匯入的檔案

在 Firestore 中查看檔案項目

觸發工作流程

查看工作狀態

查看 BigQuery 中的資料

排定工作流程

清除所用資源

刪除個別資源

刪除專案

後續步驟

使用工作流程將資料從 Cloud Storage 載入至 BigQuery