部署 Datastream 和 Dataflow 來執行數據分析

將 Datastream 與 Dataflow 工作範本整合，即可在將資料複製到 BigQuery 時，自訂串流處理邏輯，以便進行近乎即時的分析。例如，您可以轉換傳輸中的資料，或手動設定邏輯主鍵，再寫入目的地資料表。

在本教學課程中，您將使用 Datastream，將來源 MySQL 資料庫的串流資料即時插入、更新和刪除作業，串流至 Cloud Storage bucket。接著，您會為 bucket 設定 Pub/Sub 通知，以便接收新的變更檔案，並執行 Dataflow 工作，處理這些變更並合併至 BigQuery。

下圖說明整合 Datastream 和 Dataflow 的架構：

目標

在本教學課程中，您將執行下列作業：

• 在 Cloud Storage 建立 bucket。Datastream 會將來源 MySQL 資料庫的結構定義、資料表和資料，串流至這個目的地 bucket。
• 啟用 Cloud Storage bucket 的 Pub/Sub 通知。這樣一來，bucket 就會傳送通知，讓 Dataflow 知道有新檔案可供處理。這些檔案包含 Datastream 從來源資料庫串流至 bucket 的資料變更。
• 在 BigQuery 建立資料集。BigQuery 會使用資料集，存放從 Dataflow 接收的資料。這項資料代表 Datastream 串流至 Cloud Storage bucket 的來源資料庫變更。
• 為來源資料庫和 Cloud Storage 中的目的地 bucket 建立及管理連線設定檔。Datastream 中的串流會使用連線設定檔中的資訊，將資料從來源資料庫轉移至 bucket。
• 建立並啟動串流。這個串流會將來源資料庫中的資料、結構定義和資料表，轉移至 bucket。
• 確認 Datastream 將與來源資料庫結構定義相關聯的資料和資料表轉移至 bucket。
• 在 Dataflow 建立工作。Datastream 將來源資料庫的資料變更串流至 Cloud Storage bucket 後，系統會向 Dataflow 傳送通知，告知有包含變更的新檔案。Dataflow 工作會處理檔案，並將變更內容轉移至 BigQuery。
• 確認 Dataflow 處理含有這項資料相關變更的檔案，並將變更內容移轉至 BigQuery。這樣一來，Datastream 和 BigQuery 之間就能端對端整合。
• 清理您在 Datastream、Cloud Storage、Pub/Sub、Dataflow 和 BigQuery 建立的資源，以免占用配額，並避免日後產生費用。

費用

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

• Datastream
• Cloud Storage
• Pub/Sub
• Dataflow
• BigQuery

您可以使用 Pricing Calculator，根據預測用量估算費用。

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

事前準備

1. In the Google Cloud console, on the project selector page, select or create a Google Cloud project.

   Roles required to select or create a project

   • Select a project: Selecting a project doesn't require a specific IAM role—you can select any project that you've been granted a role on.
   • Create a project: To create a project, you need the Project Creator role (roles/resourcemanager.projectCreator), which contains the resourcemanager.projects.create permission. Learn how to grant roles.
   Note: If you don't plan to keep the resources that you create in this procedure, create a project instead of selecting an existing project. After you finish these steps, you can delete the project, removing all resources associated with the project.

   Go to project selector

2. Verify that billing is enabled for your Google Cloud project.

3. 啟用 Datastream API。

   啟用 API

4. 確認您的使用者帳戶具備 Datastream 管理員角色。

   前往「IAM」頁面

5. 確認您有 Datastream 可存取的來源 MySQL 資料庫。此外，請確認資料庫中有資料、資料表和結構定義。
6. 設定 MySQL 資料庫，允許來自 Datastream 公開 IP 位址的連入連線。如需所有 Datastream 區域及其相關聯公開 IP 位址的清單，請參閱「IP 許可清單和區域」。
7. 為來源資料庫設定變更資料擷取 (CDC)。詳情請參閱「設定來源 MySQL 資料庫」。

8. 確認您符合所有必要條件，可啟用 Cloud Storage 的 Pub/Sub 通知。

   在本教學課程中，您會在 Cloud Storage 建立目的地 bucket，並啟用該 bucket 的 Pub/Sub 通知。這樣一來，Dataflow 就能收到 Datastream 寫入值區的新檔案通知。這些檔案包含 Datastream 從來源資料庫串流至 bucket 的資料變更。

需求條件

Datastream 提供多種來源選項、目的地選項和網路連線方式。

在本教學課程中，我們假設您使用的是獨立 MySQL 資料庫和目的地 Cloud Storage 服務。如果是來源資料庫，您應該可以設定網路，新增傳入防火牆規則。來源資料庫可以位於地端部署環境或雲端服務供應商。如果目的地是 Cloud Storage，則不必設定連線。

由於我們無法得知您環境的具體情況，因此無法提供網路設定的詳細步驟。

在本教學課程中，請選取「IP 許可清單」做為網路連線方式。IP 許可清單是一項安全防護功能，通常用於限制及控管來源資料庫中的資料存取權，僅允許信任的使用者存取。您可以使用 IP 許可清單，建立信任的 IP 位址或 IP 範圍清單，讓使用者和其他 Google Cloud 服務 (例如 Datastream) 存取這項資料。如要使用 IP 許可清單，您必須開放來源資料庫或防火牆，允許來自 Datastream 的連線連入。

在 Cloud Storage 建立 bucket

在 Cloud Storage 建立目的地 bucket，Datastream 會將來源 MySQL 資料庫的結構定義、資料表和資料串流至該 bucket。

1. 前往 Google Cloud 控制台的 Cloud Storage「Browser」(瀏覽器) 頁面。

   前往「Browser」(瀏覽器) 頁面

2. 點選「建立 bucket」。「Create a bucket」(建立值區) 頁面隨即顯示。

3. 在「Name your bucket」(為值區命名) 區域的文字欄位中，輸入值區的專屬名稱，然後按一下「Continue」(繼續)。

4. 接受頁面其餘各區域的預設設定。在每個區域的結尾，按一下「Continue」(繼續)。

5. 點選「建立」。

為 Cloud Storage bucket 啟用 Pub/Sub 通知

在本節中，您將為建立的 Cloud Storage bucket 啟用 Pub/Sub 通知。這樣一來，只要 Datastream 將新檔案寫入 bucket，bucket 就會通知 Dataflow。這些檔案包含 Datastream 從來源 MySQL 資料庫串流至 bucket 的資料變更。

為 bucket 啟用通知前，請確認您符合所有必要條件。

1. 存取您建立的 Cloud Storage bucket。「Bucket details」(Bucket 詳細資料) 頁面隨即顯示。

2. 按一下「Activate Cloud Shell」(啟用 Cloud Shell)。

3. 在提示中輸入下列指令：

   gcloud storage buckets notifications create gs://bucket-name --topic=my_integration_notifs --payload-format=json --object-prefix=integration/tutorial/

   bucket-name 是 Cloud Storage bucket 名稱的預留位置。

   輸入這項指令後，系統會在 Pub/Sub 中建立 my_integration_notifs 主題。您也會設定 Pub/Sub 通知，只監聽您在「建立 Cloud Storage 目的地連線設定檔」步驟中定義的 Datastream 輸出路徑。 在本程序稍後的步驟中，您會在 Pub/Sub 設定這個主題，以便在 Datastream 串流至 Cloud Storage bucket 的資料發生異動時，將通知傳送至 Dataflow。

   Pub/Sub 主題會擷取這項資料的所有變更。這個主題的所有訂閱者 (例如 Dataflow) 都會收到這項資訊。

4. 選用：如果出現「Authorize Cloud Shell」視窗，請按一下「Authorize」。

5. 確認您看到下列程式碼：

   Created Cloud Pub/Sub topic projects/project-name/topics/my_integration_notifs
   Created notification config projects/_/buckets/bucket-name/notificationConfigs/1
   

   project-name 是 Google Cloud 專案名稱的預留位置。

6. 前往 Google Cloud 控制台的 Pub/Sub「Topics」(主題) 頁面。

   前往「Topics」(主題) 頁面

7. 按一下您建立的 my_integration_notifs 主題。

8. 在「my_integration_notifs」my_integration_notifs頁面中，前往頁面底部。確認「訂閱項目」分頁已啟用，且畫面上顯示「沒有可顯示的訂閱項目」訊息。

   您將為 my_integration_notifs 主題建立訂閱項目。訂閱這個訂閱項目的應用程式 (例如 Dataflow) 可以取得主題中的資訊。這項資訊與來源資料庫中的資料變更有關，Datastream 會將這些資料串流至 Cloud Storage bucket。

9. 按一下「Create Subscription」 (建立訂閱項目)。

10. 在顯示的選單中，選取「Create subscription」(建立訂閱項目)。

11. 在「將訂閱項目新增至主題」頁面中：

    1. 在「Subscription ID」(訂閱項目 ID) 欄位中輸入 my_integration_notifs_sub。
    2. 將「確認期限」值設為 120 秒。這樣 Dataflow 才有足夠的時間確認已處理的檔案，並提升 Dataflow 工作的整體效能。如要進一步瞭解 Pub/Sub 訂閱項目屬性，請參閱「訂閱項目屬性」。
    3. 頁面上的其他設定均保留預設值。
    4. 點選「建立」。

在本教學課程稍後的步驟中，您會建立 Dataflow 工作。建立這項工作時，您會將 Dataflow 指派為 my_integration_notifs_sub 訂閱項目的訂閱者。這樣一來，Dataflow 就能接收 Datastream 寫入 Cloud Storage 的新檔案通知、處理檔案，並將資料變更轉移至 BigQuery。

在 BigQuery 建立資料集

在本節中，您將在 BigQuery 建立資料集，BigQuery 會使用資料集，存放從 Dataflow 接收的資料。這項資料代表 Datastream 串流至 Cloud Storage bucket 的來源 MySQL 資料庫變更。

1. 前往 Google Cloud 控制台的 BigQuery「SQL workspace」(SQL 工作區) 頁面。

   前往「SQL workspace」(SQL 工作區) 頁面

2. 在「Explorer」窗格中，點選 Google Cloud 專案名稱旁的「查看動作」圖示 。

3. 在隨即顯示的選單中，選取「建立資料集」。

4. 在「建立資料集」視窗中：

   1. 在「Dataset ID」(資料集 ID) 欄位中，輸入資料集 ID。在本教學課程中，請在欄位中輸入 My_integration_dataset_log。
   2. 視窗中的其他預設值則維持不變。
   3. 點選「建立資料集」。
   Dataflow 會使用 My_integration_dataset_log 資料集，暫存從 Datastream 接收的資料變更。

5. 在「Explorer」窗格中，點選 Google Cloud 專案名稱旁的 「Expand node」(展開節點)，確認您建立的資料集有顯示。

6. 請按照本程序中的步驟，建立第二個資料集：My_integration_dataset_final。

   My_integration_dataset_log 資料集中暫存的變更會合併至 My_integration_dataset_final 資料集，建立來源資料庫中資料表的一對一副本。

7. 在每個資料集旁，展開 「Expand node」(展開節點)。

8. 確認每個資料集都是空的。

Datastream 將來源資料庫的資料異動串流至 Cloud Storage bucket 後，Dataflow 工作會處理含有異動內容的檔案，並將異動內容轉移至 BigQuery 資料集。

在 Datastream 建立連線設定檔

在本節中，您將在 Datastream 為來源資料庫和目的地建立連線設定檔。建立連線設定檔時，請選取「MySQL」做為來源連線設定檔類型，並選取「Cloud Storage」做為目的地連線設定檔類型。

Datastream 會使用連線設定檔中定義的資訊，連線至來源和目的地，以便將來源資料庫中的資料串流至 Cloud Storage 中的目的地 bucket。

為 MySQL 資料庫建立來源連線設定檔

1. 在 Google Cloud 控制台中，前往 Datastream 的「Connection profiles」(連線設定檔) 頁面。

   前往「Connection profiles」(連線設定檔) 頁面

2. 按一下 Create profile。

3. 如要為 MySQL 資料庫建立來源連線設定檔，請在「建立連線設定檔」頁面中，點選「MySQL」設定檔類型。

4. 在「建立 MySQL 設定檔」頁面的「定義連線設定」部分，提供下列資訊：

   • 在「Connection profile name」(連線設定檔名稱) 欄位中，輸入 My Source Connection Profile。
   • 保留自動產生的連線設定檔 ID。

   • 選取要儲存連線設定檔的「區域」。

     連線設定檔與所有資源一樣，都會儲存在區域中，且串流只能使用儲存在相同區域的連線設定檔。選取區域不會影響 Datastream 能否連線至來源或目的地，但如果區域發生停機，則可能影響可用性。

   • 輸入連線詳細資料：

     • 在「Hostname or IP」(主機名稱或 IP) 欄位，輸入 Datastream 可用來連線至來源資料庫的主機名稱或公開 IP 位址。您在本教學課程中使用 IP 許可清單做為網路連線方式，因此請提供公開 IP 位址。
     • 在「Port」(通訊埠) 欄位中，輸入為來源資料庫保留的通訊埠編號。MySQL 資料庫的預設通訊埠通常是 3306。
     • 輸入使用者名稱和密碼，向來源資料庫驗證身分。

5. 在「定義連線設定」部分，按一下「繼續」。「Create MySQL profile」(建立 MySQL 設定檔) 頁面的「Secure your connection to your source」(保護與來源的連線) 區段會啟用。

6. 在「加密類型」選單中，選取「無」。如要進一步瞭解這個選單，請參閱「建立 MySQL 資料庫的連線設定檔」。

7. 在「Secure your connection to your source」(保護與來源的連線) 區段中，按一下「Continue」(繼續)。「建立 MySQL 設定檔」頁面的「定義連線方式」部分會啟用。

8. 在「Connectivity method」(連線方式) 下拉式選單中，選擇要用於建立 Datastream 與來源資料庫之間連線的網路方法。在本教學課程中，請選取「IP 許可清單」做為連線方式。

9. 設定來源資料庫，允許從顯示的 Datastream 公開 IP 位址傳入連線。

10. 在「定義連線方式」部分，按一下「繼續」。「建立 MySQL 設定檔」頁面的「測試連線設定檔」部分會啟用。

11. 點選「執行測試」，確認來源資料庫和 Datastream 可以相互通訊。

12. 確認您看到「Test passed」(測試通過) 狀態。

    如果測試失敗，請在流程的適當部分解決問題，然後返回重新測試。如需疑難排解步驟，請參閱「診斷問題」頁面。

13. 點選「建立」。

建立 Cloud Storage 目的地連線設定檔

1. 在 Google Cloud 控制台中，前往 Datastream 的「Connection profiles」(連線設定檔) 頁面。

   前往「Connection profiles」(連線設定檔) 頁面

2. 按一下 Create profile。

3. 如要為 Cloud Storage 建立目的地連線設定檔，請在「Create a connection profile」(建立連線設定檔) 頁面中，點選「Cloud Storage」設定檔類型。

4. 在「Create Cloud Storage profile」(建立 Cloud Storage 設定檔) 頁面中，提供下列資訊：

   • 在「Connection profile name」(連線設定檔名稱) 欄位中，輸入 My Destination Connection Profile。
   • 保留自動產生的連線設定檔 ID。
   • 選取要儲存連線設定檔的「區域」。

   • 在「Connection details」(連線詳細資料) 窗格中，按一下「Browse」(瀏覽)，選取稍早在本教學課程中建立的 Cloud Storage bucket。Datastream 會將來源資料庫中的資料轉移至這個 bucket。選取完畢後，按一下「選取」。

     「Connection details」(連線詳細資料) 窗格的「Bucket name」(值區名稱) 欄位會顯示您的值區。

   • 在「Connection profile path prefix」(連線設定檔路徑前置字串) 欄位中，提供要附加至 bucket 名稱的路徑前置字串，以便 Datastream 將資料串流至目的地。請確認 Datastream 將資料寫入 bucket 內的路徑，而非 bucket 根資料夾。在本教學課程中，請使用您在設定 Pub/Sub 通知時定義的路徑。在欄位中輸入 /integration/tutorial。

5. 點選「建立」。

為 MySQL 資料庫建立來源連線設定檔，並為 Cloud Storage 建立目的地連線設定檔後，即可使用這些設定檔建立串流。

在 Datastream 建立串流

在這個部分，您要建立串流。這項串流會使用連線設定檔中的資訊，將資料從來源 MySQL 資料庫移轉至 Cloud Storage 中的目的地 bucket。

Datastream 內建設定輔助程式，可協助您建立串流。這個助理包含六個面板：「開始使用」、「定義及測試來源」、「設定來源」、「定義目的地」、「設定目的地」和「檢查及建立」。如要填入各面板，請按照下列各節的指示操作。

定義串流的設定

1. 在 Google Cloud 控制台中，前往 Datastream 的「Streams」(串流) 頁面。

   前往「Streams」(串流) 頁面

2. 按一下 [建立訊息串]。

3. 在「建立串流」頁面的「定義串流詳細資料」面板中，提供下列資訊：

   • 在「Stream name」(串流名稱) 欄位中輸入 My Stream。
   • 保留自動產生的串流 ID。
   • 在「Region」(區域) 選單中，選取您建立來源和目的地連線設定檔的區域。
   • 在「Source type」(來源類型) 選單中，選取「MySQL」設定檔類型。
   • 在「Destination type」(目的地類型) 選單中，選取「Cloud Storage」設定檔類型。

4. 查看自動產生的必要條件，瞭解如何為串流準備環境。包括如何設定來源資料庫，以及如何將 Datastream 連結至 Cloud Storage 中的目的地 bucket。

   如要進一步瞭解這些必要條件，請參閱「設定來源 MySQL 資料庫」。

5. 按一下「繼續」。「建立串流」頁面的「定義 MySQL 連線設定檔」面板隨即顯示。

指定來源連線設定檔的相關資訊

在這個部分，您要選取為來源資料庫建立的連線設定檔 (來源連線設定檔)。在本教學課程中，請選取「My Source Connection Profile」。

1. 在「來源連線設定檔」選單中，選取 MySQL 資料庫的來源連線設定檔。

2. 點選「執行測試」，確認來源資料庫和 Datastream 可以相互通訊。

   如果測試失敗，系統會顯示與連線設定檔相關聯的問題。如需疑難排解步驟，請參閱「診斷問題」頁面。進行必要變更來修正問題，然後重新測試。

3. 按一下「繼續」。「Create stream」(建立串流) 頁面的「Configure stream source」(設定串流來源) 面板隨即顯示。

設定串流的來源資料庫相關資訊

在本節中，您將指定來源資料庫中的資料表和結構定義，設定串流的來源資料庫相關資訊，讓 Datastream：

• 可轉移至目的地。
• 無法轉移至目的地。

您也可以決定 Datastream 是否要補充歷來資料，以及是否要將持續變更串流至目的地，或只串流資料變更。

1. 使用「Objects to include」(要包含的物件) 選單，指定 Datastream 可轉移至 Cloud Storage 目的地 bucket 資料夾的來源資料庫資料表和結構定義。只有在資料庫最多有 5,000 個物件時，系統才會載入選單。

   在本教學課程中，Datastream 應移轉所有資料表和結構定義。因此，請從選單中選取「All tables from all schemas」(採用任何結構定義的所有資料表)。

2. 確認「選取要排除的物件」面板已設為「無」。您不希望限制 Datastream 將來源資料庫中的任何資料表和結構定義移轉至 Cloud Storage。

3. 確認「Choose backfill mode for historical data」(選擇歷來資料的補充作業模式) 面板設為「Automatic」(自動)。Datastream 會將所有現有資料和資料變更，從來源串流至目的地。

4. 按一下「繼續」。「Create stream」(建立串流) 頁面的「Define Cloud Storage connection profile」(定義 Cloud Storage 連線設定檔) 面板隨即顯示。

選取目的地連線設定檔

在本節中，您將選取為 Cloud Storage 建立的連線設定檔 (目的地連線設定檔)。在本教學課程中，請輸入「My Destination Connection Profile」。

1. 在「Destination connection profile」(目的地連線設定檔) 選單中，選取 Cloud Storage 的目的地連線設定檔。

2. 按一下「繼續」。「Create stream」(建立串流) 頁面的「Configure stream destination」(設定串流目的地) 面板隨即顯示。

設定串流目的地的相關資訊

在本節中，您將設定串流的目的地 bucket 相關資訊。這類資訊包括：

• 寫入 Cloud Storage 的檔案輸出格式。
• 目的地 bucket 的資料夾，Datastream 會將來源資料庫的結構定義、資料表和資料移轉至此。

1. 在「Output format」(輸出格式) 欄位中，選取寫入 Cloud Storage 的檔案格式。Datastream 支援兩種輸出格式：Avro 和 JSON。在本教學課程中，檔案格式為 Avro。

   您不必填寫「Stream path prefix」(串流路徑前置字元) 欄位。在這個欄位中，您可以為路徑提供前置字元，當 Datastream 將資料移轉至目的地時，這個前置字元會附加至 bucket 名稱。Datastream 會將來源 MySQL 資料庫的結構定義、資料表和資料，移轉至這個 Cloud Storage bucket 路徑。

   您在建立 Cloud Storage 目的地連線設定檔時，已提供 /integration/tutorial 路徑。

2. 按一下「繼續」。「Create stream」(建立串流) 頁面的「Review stream details and create」(查看串流詳細資料並建立) 面板隨即顯示。

建立串流

1. 確認串流的詳細資料，以及串流用於將資料從來源 MySQL 資料庫移轉至 Cloud Storage 目的地 bucket 的來源和目的地連線設定檔。

2. 如要驗證串流，請按一下「Run validation」(執行驗證)。Datastream 會驗證串流，確認來源設定正確無誤、串流可連至來源和目的地，以及串流的端對端設定正確。

   如果通過驗證檢查，系統會顯示勾號圖示。

   如果驗證檢查未通過，系統會顯示驚嘆號圖示和「查看錯誤詳細資料」按鈕。點選按鈕後，系統會顯示對話方塊，說明檢查未通過的原因，以及修正問題的方法。完成適當修正後，按一下「重新驗證」。

   如要進一步瞭解如何排解未通過驗證檢查的問題，請參閱「診斷問題」。

3. 通過所有驗證檢查後，按一下「Create」(建立)。

4. 在「要建立串流嗎？」對話方塊中，按一下「建立」。

開始串流

在本教學課程中，您會分別建立及啟動串流，以免串流建立程序增加來源資料庫的負載。如要延後負載，請建立串流但不啟動，等到資料庫可以處理負載時再啟動。

啟動串流後，Datastream 就能將來源資料庫中的資料、結構定義和資料表轉移至目的地。

1. 在 Google Cloud 控制台中，前往 Datastream 的「Streams」(串流) 頁面。

   前往「Streams」(串流) 頁面

2. 找出要啟動的串流，然後勾選旁邊的核取方塊。在本教學課程中，請輸入「My Stream」。

3. 按一下「啟動」。

4. 在對話方塊中，按一下「Start」。串流狀態會從「Not started」變更為「Starting」，再變更為「Running」。

   串流大約需要 30 秒才會開始執行。背景資源必須啟動，直播才能開始。

啟動串流後，您可以確認 Datastream 是否將資料從來源資料庫轉移至目的地。

驗證串流

在本節中，您將確認 Datastream 將來源 MySQL 資料庫中所有資料表的資料，轉移至 Cloud Storage 目的地 bucket 的 /integration/tutorial 資料夾。

1. 在 Google Cloud 控制台中，前往 Datastream 的「Streams」(串流) 頁面。

   前往「Streams」(串流) 頁面

2. 按一下您建立的串流。在本教學課程中，請輸入 My Stream。

3. 在「Stream details」(串流詳細資料) 頁面中，按一下 bucket-name/integration/tutorial 連結，其中 bucket-name 是您為 Cloud Storage bucket 指定的名稱。這個連結會顯示在「Destination write path」(目的地寫入路徑) 欄位後方。Cloud Storage 的「Bucket details」(bucket 詳細資料) 頁面會在另一個分頁開啟。

   連結是由 bucket 名稱和 bucket 資料夾組成，Datastream 會將來源資料庫的結構定義、資料表和資料轉移至該資料夾。您在建立 Cloud Storage 的目的地連線設定檔時，已將這個資料夾指定為 /integration/tutorial。

4. 確認您看見代表來源資料庫資料表的資料夾。

5. 按一下其中一個資料表資料夾，然後逐一按一下各個子資料夾，直到看到與資料表相關聯的資料。

   第一個資料夾是 [schema]_[table]，後續資料夾則代表 Datastream 將資料從來源資料庫轉移至 Cloud Storage 目的地 bucket 的年、月、日、時和分。

   系統每分鐘會建立一個資料夾 (如有新資料要寫入)。

   當檔案大小達到 250 MB 或結構定義變更時，系統會建立新檔案。如果資料表已分區，系統會為每個分區建立檔案。

建立 Dataflow 工作

在本節中，您會在 Dataflow 建立工作。Datastream 將來源 MySQL 資料庫的資料變更串流至 Cloud Storage bucket 後，Pub/Sub 會向 Dataflow 傳送通知，告知有包含變更的新檔案。Dataflow 工作會處理檔案，並將變更內容轉移至 BigQuery。

1. 前往 Google Cloud 控制台的 Dataflow「Jobs」(工作) 頁面。

   前往「Jobs」(工作) 頁面

2. 點選「依據範本建立工作」。

3. 在「Create job from template」(依據範本建立工作) 頁面的「Job name」(工作名稱) 欄位中，輸入要建立的 Dataflow 工作名稱。在本教學課程中，請在欄位中輸入 my-dataflow-integration-job。

4. 從「Regional endpoint」(區域端點) 選單中，選取要儲存工作的區域。這個區域與您建立的來源連線設定檔、目的地連線設定檔和串流所選的區域相同。

5. 從「Dataflow template」(Dataflow 範本) 選單中，選取要用來建立工作的範本。在本教學課程中，請選取「Datastream to BigQuery」。

   選取後，系統會顯示與這個範本相關的其他欄位。

   「Datastream 到 BigQuery」範本是串流管道，可讀取 Datastream 資料並複製到 BigQuery。範本會使用 Pub/Sub 通知從 Cloud Storage 讀取資料，並將資料複製到依時間分區的 BigQuery 暫存資料表。複製完成後，範本會在 BigQuery 中執行 Merge，將所有變更資料擷取 (CDC) 變更 upsert 至來源資料表的副本。

   為盡量降低頻繁執行 Merge 作業的相關費用，建議您一開始先設定每 6 到 12 小時執行一次。所有補充作業完成且資料順利複製後，請將這個值降低至所選頻率。

   如要進一步瞭解「Datastream 至 BigQuery」範本，請參閱「Datastream 至 BigQuery (串流)」。

6. 在「File location for Datastream file output in Cloud Storage.」(Cloud Storage 中 Datastream 檔案輸出的檔案位置。) 欄位中，使用下列格式輸入 Cloud Storage bucket 的名稱：gs://bucket-name。

7. 在「Pub/Sub subscription being used in a Cloud Storage notification policy.」(Cloud Storage 通知政策中使用的 Pub/Sub 訂閱項目。) 欄位中，輸入包含 Pub/Sub 訂閱項目名稱的路徑。在本教學課程中，請輸入 projects/project-name/subscriptions/my_integration_notifs_sub。

   project-name 是 Google Cloud 專案名稱的預留位置。此外，您在本教學課程的「為 Cloud Storage bucket 啟用 Pub/Sub 通知」一節中，建立了 my_integration_notifs_sub 訂閱項目。

8. 在本教學課程中，Datastream 會以 Avro 格式將檔案寫入 Cloud Storage，因此請在「Datastream output file format (avro/json).」(Datastream 輸出檔案格式 (avro/json)) 欄位中輸入 avro。

9. 在「Name or template for the dataset to contain staging tables.」(要包含暫存資料表的資料集名稱或範本。) 欄位中輸入 My_integration_dataset_log，因為 Dataflow 會使用這個資料集，暫存從 Datastream 接收的資料變更。

10. 在「Template for the dataset to contain replica tables.」(要包含副本資料表的資料集範本。)欄位中輸入 My_integration_dataset_final，因為這個資料集會合併 My_integration_dataset_log 資料集中的暫存變更，建立來源資料庫資料表的一對一副本。

    您已在本教學課程的「在 BigQuery 建立資料集」部分，建立 My_integration_dataset_log 和 My_integration_dataset_final 資料集。

11. 在「Dead letter queue directory」(無效信件佇列目錄) 欄位中，輸入路徑，其中包含獨立 Cloud Storage bucket 的名稱，以及無效信件佇列的資料夾。請勿使用根資料夾中的路徑，且路徑不得與 Datastream 寫入資料的路徑相同。Dataflow 無法轉移至 BigQuery 的資料變更會儲存在佇列中。您可以修正佇列中的內容，讓 Dataflow 重新處理。

    在本教學課程中，請在「Dead letter queue directory」(無效信件佇列目錄) 欄位中輸入 gs://dlq-bucket-name/dlq (其中 dlq-bucket-name 是專用於無效信件佇列的 Cloud Storage bucket 名稱，dlq 則是無效信件佇列的資料夾)。

12. 按一下「Run Job」(執行工作)。

Dataflow 工作可能需要幾分鐘才能執行。背景資源必須執行。此外，工作必須先處理從 Cloud Storage 接收的資料變更檔案，才能將這些變更移轉至 BigQuery。

驗證整合設定

在本教學課程的「驗證串流」部分，您已確認 Datastream 將來源 MySQL 資料庫中所有資料表的資料，轉移至 Cloud Storage 目的地 bucket 的 /integration/tutorial 資料夾。

在本節中，您將確認 Dataflow 處理了包含與這項資料相關變更的檔案，並將變更轉移至 BigQuery。這樣一來，Datastream 和 BigQuery 之間就能端對端整合。

1. 前往 Google Cloud 控制台的 BigQuery「SQL workspace」(SQL 工作區) 頁面。

   前往「SQL workspace」(SQL 工作區) 頁面

2. 在「Explorer」窗格中，展開 Google Cloud 專案名稱旁的節點。

3. 展開 My_integration_dataset_log 和 My_integration_dataset_final 資料集旁的節點。

4. 確認每個資料集現在都包含資料。這表示 Dataflow 已處理檔案，其中包含與 Datastream 串流至 Cloud Storage 的資料相關聯的變更，並將這些變更轉移至 BigQuery。

清除所用資源

為避免系統向您的 Google Cloud 帳戶收取本教學課程所用資源的費用，請使用 Google Cloud 控制台執行下列操作：

• 刪除專案、Datastream 串流和 Datastream 連線設定檔。
• 停止 Dataflow 工作。
• 刪除 BigQuery 資料集、Pub/Sub 主題和訂閱項目，以及 Cloud Storage bucket。

清理在 Datastream、Dataflow、BigQuery、Pub/Sub 和 Cloud Storage 建立的資源，可避免資源占用配額，也不會產生後續費用。

刪除專案

如要避免付費，最簡單的方法就是刪除您為了本教學課程所建立的專案。

刪除專案會出現以下結果：

• 專案中的所有內容都會遭到刪除。如果您使用現有專案來進行本教學課程，刪除專案將一併刪除其中已完成的所有工作。
• 自訂專案 ID 會消失。您建立這個專案時，可能已建立日後使用的自訂專案 ID。如要保留使用該專案 ID 的網址 (例如 appspot.com 網址)，請刪除在該專案中選取的資源，而不是刪除整個專案。

如果打算進行多個教學課程及快速入門導覽課程，重複使用專案有助於避免超出專案配額限制。

1. 前往 Google Cloud 控制台的「Manage resources」(管理資源) 頁面。

   前往「管理資源」頁面

2. 在專案清單中選取要刪除的專案，然後點按「刪除」。

3. 在對話方塊中輸入專案 ID，然後按一下「Shut down」(關閉)，即可刪除專案。

刪除串流

1. 在 Google Cloud 控制台中，前往 Datastream 的「Streams」(串流) 頁面。

   前往「Streams」(串流) 頁面

2. 按一下要刪除的串流。在本教學課程中，請輸入「My Stream」。

3. 點選「暫停」。

4. 在對話方塊中，按一下「暫停」。

5. 在「Stream details」(串流詳細資料) 頁面的「Stream status」(串流狀態) 窗格中，確認串流狀態為 Paused。

   您先暫停串流再刪除，確保所有傳輸中的資料都從來源資料庫轉移至目的地。

6. 點選「刪除」。

7. 在對話方塊的文字欄位中輸入 Delete，然後按一下「Delete」(刪除)。

刪除連線設定檔

1. 在 Google Cloud 控制台中，前往 Datastream 的「Connection profiles」(連線設定檔) 頁面。

   前往「Connection profiles」(連線設定檔) 頁面

2. 勾選要刪除的連線設定檔核取方塊：「My Source Connection Profile」和「My Destination Connection Profile」。

3. 點選「刪除」。

4. 在對話方塊中，按一下「刪除」。

停止 Dataflow 工作

您無法在 Dataflow 中刪除工作。不過，您可以停止工作，讓工作不再處理含有資料異動的檔案 (這些檔案會串流至 Cloud Storage bucket)，或將資料異動移轉至 BigQuery。

1. 前往 Google Cloud 控制台的 Dataflow「Jobs」(工作) 頁面。

   前往「Jobs」(工作) 頁面

2. 按一下要停止的工作。在本教學課程中，這是 my-dataflow-integration-job。

3. 按一下「停止」。

4. 在「Stop job」(停止工作) 對話方塊中，選取「Drain」(排空) 選項，然後按一下「Stop job」(停止工作)。

   Dataflow 會處理完所有剩餘的緩衝資料，確保所有「傳輸中」的資料都移轉至 BigQuery，然後才停止工作。

刪除 BigQuery 資料集

1. 前往 Google Cloud 控制台的 BigQuery「SQL workspace」(SQL 工作區) 頁面。

   前往「SQL workspace」(SQL 工作區) 頁面

2. 在「Explorer」窗格中，展開 Google Cloud 專案名稱旁的節點。

3. 在「在 BigQuery 建立資料集」部分，您建立了一些資料集。請找到其中一個資料集，然後點選右側的「查看動作」按鈕。這個按鈕看起來像垂直刪節號。

   在本教學課程中，請按一下「My_integration_dataset_log」右側的「查看動作」按鈕。

4. 在顯示的下拉式選單中，選取「刪除」。

5. 在「Delete dataset?」(要刪除資料集嗎？) 對話方塊中，在文字欄位輸入 delete，然後按一下「Delete」(刪除)。

6. 重複執行本程序中的步驟，刪除您建立的第二個資料集：My_integration_dataset_final。

刪除 Pub/Sub 訂閱項目和主題

1. 前往 Google Cloud 控制台的 Pub/Sub「Subscriptions」(訂閱項目) 頁面。

   前往「訂閱項目」頁面

2. 找出要刪除的訂閱項目，然後勾選旁邊的核取方塊。在本教學課程中，請勾選 my_integration_notifs_sub 訂閱項目旁的核取方塊。

3. 點選「刪除」。

4. 在「Delete subscription」(刪除訂閱項目) 對話方塊中，點選「Delete」(刪除)。

5. 前往 Google Cloud 控制台的 Pub/Sub「Topics」(主題) 頁面。

   前往「Topics」(主題) 頁面

6. 勾選「my_integration_notifs」my_integration_notifs主題旁的核取方塊。

7. 點選「刪除」。

8. 在「Delete topic」(刪除主題) 對話方塊中，在文字欄位輸入 delete，然後按一下「Delete」(刪除)。

刪除 Cloud Storage bucket

1. 前往 Google Cloud 控制台的 Cloud Storage「Browser」(瀏覽器) 頁面。

   前往「Browser」(瀏覽器) 頁面

2. 勾選 bucket 旁的核取方塊。

3. 點選「刪除」。

4. 在對話方塊的文字欄位中輸入 Delete，然後按一下「Delete」(刪除)。

後續步驟

• 進一步瞭解 Datastream。
• 使用舊版串流 API，執行進階功能，將資料串流至 BigQuery。
• 歡迎自行試用其他 Google Cloud 功能，並參考我們的教學課程。