Datastream 和 Dataflow 進行分析MySQL以CDC的方式同步到BigQuery

作者：MeshCloud脈時雲 公有雲架構師 於文寶

背景

對於擁有許多獨立數據源的企業而言，訪問整個組織內的企業數據（尤其是實時訪問）並非易事。這會導致數據訪問受限且速度緩慢，因而造成組織無法進行檢查。

Datastream 提供近乎實時的訪問權限，讓您能夠更改各種本地和雲端數據源中的數據以創建組織數據訪問權限。Datastream 提供簡單的設置體驗和統一的使用 API，讓組織普遍能夠訪問組織內可用的最新企業數據，從而為集成式近乎實時的場景提供支持。

其中一種場景是，將數據從源數據庫轉移到雲端存儲服務或消息傳遞隊列，然後將這些數據轉換為可供與該存儲服務或消息傳遞隊列通信的其他應用和服務（例如 Dataflow）讀取的形式。Dataflow 是一項用於在 Google Cloud 上捕獲和處理數據的 Web 服務。

在本教程中，您將了解 Datastream 如何通過簡化的 Dataflow 模板與 Dataflow 無縫集成，從而在 BigQuery 中實現最新具體化視圖以執行分析。

您將了解如何使用 Dataflow 將更改（插入、更新或刪除的數據）從源 MySQL 數據庫流式傳輸到 Cloud Storage 存儲桶中的文件夾。

您將配置 Cloud Storage 存儲桶以發送通知，供 Dataflow 了解包含 Datastream 從源數據庫流式傳輸的數據更改的任何新文件。然後，Dataflow 作業將處理這些文件並將更改轉移到 BigQuery。

架構圖

準備工作

1. 啟用 Datastream API。
2. 確保您已為您的用戶帳號分配 Datastream Admin 角色。
3. 確保您有一個 Datastream 可以訪問的源數據庫。本教程使用 MySQL 數據庫作為來源。
4. 確保源數據庫中有數據、表和架構。
5. 配置源數據庫以允許來自 Datastream 公共 IP 地址的傳入連接。如需查看包含所有 Datastream 地區及其關聯公共 IP 地址的列表，請參閱 IP 許可名單和地區。
6. 為源數據庫設置變更數據捕獲 (CDC)。如需了解詳情，請參閱配置源 MySQL 數據庫。
7. 確保您已配置 Datastream 可以訪問的目標 Cloud Storage 存儲桶。
8. 確保您滿足所有前提條件，以便為 Cloud Storage 啟用 Pub/Sub 通知。
9. 您將在 Cloud Storage 中創建目標存儲桶，並為存儲桶啟用 Pub/Sub 通知。這樣設置後，Dataflow 就可以接收通知來了解 Datastream 寫入存儲桶的新文件。這些文件包含 Dataflow 從源數據庫流式傳輸到存儲桶的數據更改。

要求

Datastream 提供各種來源選項、目標選項和網絡連接方法。

在本教程中，假設您使用獨立的 MySQL 數據庫和目標 Cloud Storage 服務。對於源數據庫，您應該能夠將網絡配置為添加入站防火牆規則。源數據庫可以位於本地，也可以位於雲提供商。對於 Cloud Storage 目標位置，無需配置連接。

由於我們無法獲知您環境的具體細節，因此我們無法提供網絡配置的詳細步驟。

在本教程中，您將選擇 IP 許可名單作為網絡連接方法。IP 許可名單是一項安全功能，通常用於僅限受信任的用戶訪問您的源數據庫中的數據並對這些訪問進行控制。您可以使用 IP 許可名單創建受信任的 IP 地址或 IP 地址範圍列表，您的用戶和其他 Google Cloud 服務（如 Datastream）可通過這些地址訪問此數據。要使用 IP 許可名單，您必須向來自 Datastream 的傳入連接開放源數據庫或防火牆。

在 Cloud Storage 中創建存儲桶

在本部分中，您將在 Cloud Storage 中創建存儲桶。DataStream 將架構、表和數據從源 MySQL 數據庫流式傳輸到的目標存儲桶。

1. 轉到 Google Cloud Console 中 Cloud Storage 的瀏覽器頁面。
2. 點擊創建存儲桶。此時將顯示創建存儲桶頁面。
3. 在為存儲桶命名區域的文本字段中，輸入 my-integration-bucket-001，然後點擊繼續。
4. 對於頁面每個剩餘區域，請接受默認設置。為此，請點擊每個區域底部的繼續。
5. 點擊創建。

為 Cloud Storage 存儲桶啟用 Pub/Sub 通知

在本部分中，您將為創建的 Cloud Storage 存儲桶啟用 Pub/Sub 通知。這樣，您就可以配置存儲桶來發送通知，供 Dataflow 用來了解 Datastream 寫入該存儲桶的任何新文件。這些文件包含 Datastream 從源 MySQL 數據庫流式傳輸到存儲桶的數據的更改。

在為存儲桶啟用通知之前，請確保滿足所有前提條件。

1. 訪問您創建的 Cloud Storage 存儲桶。此時將顯示存儲桶詳情頁面。
2. 點擊頁面右上角的激活 Cloud Shell 按鈕。
3. 在提示符處，輸入以下命令：
4. gsutil notification create -t my_integration_notifs -f json gs://my-integration-bucket-001
5. 通過輸入此命令，您將在 Pub/Sub 中創建 my_integration_notifs 主題。本主題將配置為向 Dataflow 發送通知，供其了解 Datastream 流式傳輸到 Cloud Storage 存儲桶 (my-integration-bucket-001) 的數據的更改。
6. 對此數據所做的任何更改都將捕獲到 Pub/Sub 主題中。此主題的任何訂閱者（例如 Dataflow）都可以獲取此信息。
7. 可選。如果看到授權 Cloud Shell 窗口，請點擊授權。
8. 驗證您是否看到以下代碼行：

1. project-name 是您的 Google Cloud 項目名稱的佔位符。
2. 轉到 Google Cloud Console 中 Pub/Sub 的主題頁面。

1. 點擊在此過程中創建的 my_integration_notifs 主題。
2. 在 my_integration_notifs 頁面，滾動到頁面底部。訂閱標籤頁處於活躍狀態。此外，會出現沒有可顯示的訂閱消息。
3. 您將為 my_integration_notifs 主題創建訂閱。訂閱此訂閱的應用（例如 Dataflow）可以獲取主題中的信息。此信息與 Datastream 流式傳輸到 Cloud Storage 存儲桶的源數據庫數據更改相關聯。
4. 點擊創建訂閱，然後從顯示的下拉菜單中選擇創建訂閱項。
5. 填充向主題添加訂閱頁面，如下所示：

1. 在訂閱 ID 字段中，輸入訂閱的 ID。在本教程中，在字段中輸入 my_integration_notifs_sub。
2. 保留頁面上的所有其他默認值。
3. 點擊創建。

在本教程後面部分，您將創建 Dataflow 作業。在創建此作業時，您需要將 Dataflow 指定為 my_integration_notifs_sub 訂閱的訂閱者。指定後，Dataflow 可以接收通知來了解 Datastream 寫入 Cloud Storage 的新文件、處理這些文件並將數據更改轉移到 BigQuery。

在 BigQuery 中創建數據集

在本部分中，您將在 BigQuery 中創建數據集。BigQuery 使用數據集來包含它從 Dataflow 接收的數據。此數據表示 Datastream 將流式傳輸到 Cloud Storage 存儲桶的源 MySQL 數據庫中的更改。

1. 轉到 Google Cloud Console 中 BigQuery 的 SQL 工作區頁面。
2. 在探索器窗格中，點擊您的 Google Cloud 項目名稱右側的查看操作按鈕。該按鈕形似垂直省略號。
3. 從顯示的下拉菜單中選擇創建數據集。
4. 填充創建數據集窗口，如下所示：

1. 在數據集 ID 字段中，輸入數據集的 ID。在本教程中，在字段中輸入 My_integration_dataset_log。
2. 保留窗口中的所有其他默認值。
3. 點擊創建數據集。

1. Dataflow 將使用 My_integration_dataset_log 數據集來暫存從 Datastream 接收的數據更改。
2. 在探索器窗格中，點擊 Google Cloud 項目名稱左側的節點圖標，並驗證您能看到自己創建的數據集。
3. 按照此過程中的步驟創建第二個數據集：My_integration_dataset_final。
4. 在 My_integration_dataset_final 數據集中，將合併在 My_integration_dataset_log 數據集中暫存的更改，以在源數據庫中創建表的一對一副本。
5. 展開每個數據集左側的節點。
6. 驗證每個數據集是否為空。

在 Datastream 將數據更改從源數據庫流式傳輸到 Cloud Storage 存儲桶後，Dataflow 作業將處理包含更改的文件，並將更改轉移到 BigQuery 數據集。

在 Datastream 中創建連接配置文件

在本部分中，您將在 Datastream 中為源數據庫和目標位置創建連接配置文件。在創建連接配置文件時，您需要選擇 MySQL 作為來源連接配置文件的類型，選擇 Cloud Storage 作為目標連接配置文件的類型。

Dataflow 使用連接配置文件中定義的信息同時連接到來源和目標位置，以便可將數據從源數據庫流式傳輸到 Cloud Storage 中的目標存儲桶。

為 MySQL 數據庫創建來源連接配置文件

1. 轉到 Google Cloud Console 中 Datastream 的連接配置文件頁面。
2. 點擊創建配置文件。
3. 在創建連接配置文件頁面中，點擊 MySQL 配置文件類型（因為您希望為 MySQL 數據庫創建來源連接配置文件）。
4. 在創建 MySQL 配置文件頁面的定義連接設置部分中提供以下信息：

• 輸入 My Source Connection Profile 作為源數據庫的連接配置文件名稱。
• 保留自動生成的連接配置文件 ID。
• 選擇用於存儲連接配置文件的地區。
• 與所有資源一樣，連接配置文件也保存在地區中，並且數據流只能使用與該數據流存儲在同一地區的連接配置文件。地區選擇不會影響 Datastream 是否可連接到來源或目標位置，但如果該地區發生停機，則可能會影響可用性。
• 輸入連接詳情：
• 在主機名或 IP 字段中，輸入 Datastream 可用於連接到源數據庫的主機名或公共 IP 地址。您將提供公共 IP 地址，因為 IP 許可名單將用作本教程的網絡連接方法。
• 在端口字段中，輸入為源數據庫預留的端口號。對於 MySQL 數據庫，默認端口通常為 3306。
• 輸入用戶名和密碼，對源數據庫進行身份驗證。

1. 在定義連接設置部分中，點擊繼續。創建 MySQL 配置文件頁面的確保與源的連接安全部分處於活躍狀態。
2. 從加密類型菜單中，選擇無。如需詳細了解此菜單，請參閱為 MySQL 數據庫創建連接配置文件。
3. 在確保與源的連接安全部分中，點擊繼續。創建 MySQL 配置文件頁面的定義連接方法部分處於活躍狀態。
4. 選擇您要在源數據庫與 Cloud Storage 中的目標存儲桶之間建立連接時使用的網絡方法。在本教程中，請使用連接方法下拉列表選擇 IP 許可名單作為網絡方法。
5. 配置源數據庫以允許來自顯示的 Datastream 公共 IP 地址的傳入連接。
6. 在定義連接方法部分中，點擊繼續。創建 MySQL 配置文件頁面的測試連接配置文件部分處於活躍狀態。
7. 點擊運行測試，驗證源數據庫和 DataStream 是否可以相互通信。
8. 驗證您是否看到「已通過測試」狀態。
9. 如果測試失敗，您可以在流的適當部分解決問題，然後重新測試。

1. 點擊創建。

為 Cloud Storage 創建目標連接配置文件

1. 轉到 Google Cloud Console 中 Datastream 的連接配置文件頁面。
2. 點擊創建配置文件。
3. 在創建連接配置文件頁面中，點擊 Cloud Storage 配置文件類型（因為您希望為 Cloud Storage 創建目標連接配置文件）。
4. 在創建 Cloud Storage 配置文件頁面中提供以下信息：

• 輸入 My Destination Connection Profile 作為目標 Cloud Storage 服務的連接配置文件名稱。
• 保留自動生成的連接配置文件 ID。
• 選擇用於存儲連接配置文件的地區。
• 在連接詳情窗格中，點擊瀏覽以選擇您在本教程前面部分創建的 my-integration-bucket-001。Datastream 會將數據從源數據庫轉移到該存儲桶中。完成選擇後，點擊選擇。
• 您的存儲桶會顯示在連接詳情窗格的存儲桶名稱字段中。
• 在連接配置文件路徑前綴字段中，提供當 Datastream 將數據流式傳輸到目標位置時要追加到存儲桶名稱的路徑的前綴。在本教程中，在字段中輸入 /integration/tutorial。
• 注意：您在此字段中輸入的任何路徑都必須以斜杠 (/) 開頭。

1. 點擊創建。

為 MySQL 數據庫創建來源連接配置文件，並為 Cloud Storage 創建目標連接配置文件後，您可以使用它們來創建數據流。

在 Datastream 中創建數據流

在本部分中，您將創建一個數據流。此數據流會將源 MySQL 數據庫中的數據轉移到 Cloud Storage 中的目標存儲桶。

創建數據流涉及到：

• 定義數據流的設置。
• 選擇您為源數據庫創建的連接配置文件（來源連接配置文件）。在本教程中，我們使用的是我的來源連接配置文件。
• 通過在源數據庫中指定 Datastream 對其執行以下操作的表和架構，配置數據流的源數據庫的相關信息：
• 可以轉移到目標位置。
• 無法轉移到目標位置。
• 確定 Datastream 是回填歷史數據並將進行中的更改流式傳輸到目標位置，還是僅流式傳輸對數據的更改。
• 選擇您為 Cloud Storage 創建的連接配置文件（目標連接配置文件）。在本教程中，我們使用的是我的目標連接配置文件。
• 配置有關數據流的目標存儲桶的信息。此類信息包括：
• DataStream 將架構、表和數據從來源數據庫轉移到的目標存儲桶的文件夾。
• 寫入 Cloud Storage 的文件的輸出格式。Datastream 目前支持兩種輸出格式：Avro 和 JSON。在本教程中，使用 Avro 文件格式。

Dataflow 配備了嚮導，可幫助您創建數據流。此嚮導包含六個面板：使用入門、定義和測試來源、配置來源、定義目標位置、配置目標位置以及審核並創建。本頁面的各個部分介紹了如何填充每個面板。

定義數據流的設置

1. 轉到 Google Cloud Console 中數據流頁面。
2. 在創建數據流頁面的定義數據流詳情面板中提供以下信息：

• 輸入 My Stream 作為數據流名稱。
• 保留自動生成的數據流 ID。
• 從地區菜單中，選擇您在其中創建了來源和目標連接配置文件的地區。

• 從來源類型菜單中，選擇 MySQL 配置文件類型。
• 從目標類型菜單中，選擇 Cloud Storage 配置文件類型。

選擇目標連接配置文件

1. 從目標連接配置文件菜單中，選擇 Cloud Storage 的目標連接配置文件。
2. 點擊繼續。系統會顯示創建數據流頁面的配置數據流目標位置面板。

配置有關數據流目標位置的信息

1. 在輸出格式字段中，選擇寫入 Cloud Storage 的文件的格式。在本教程中，使用 Avro 文件格式。
2. 在數據流路徑前綴字段中，您可以提供一個在 Datastream 將數據轉移到目標位置時要追加到存儲桶名稱中的路徑的前綴。這是您的 Cloud Storage 存儲桶的路徑，DataStream 會將架構、表和數據從源 MySQL 數據庫轉移到該存儲桶中。
3. 您在為 Cloud Storage 創建目標連接配置文件時，已提供 /integration/tutorial 路徑。因此，您無需填充此字段。
4. 點擊繼續。系統會顯示創建數據流頁面的審核數據流詳情並創建面板。

創建數據流

1. 驗證數據流的詳細信息，以及數據流用於將數據從源 MySQL 數據庫轉移到 Cloud Storage 中目標存儲桶的來源和目標連接配置文件。
2. 點擊運行驗證來驗證數據流。通過驗證數據流，Datastream 會檢查來源是否配置正確，驗證數據流是否可以連接到來源和目標位置，並驗證數據流的端到端配置。
3. 如果驗證檢查通過，則驗證檢查的左側會顯示一個對勾標記圖標。
4. 如果驗證檢查未通過，則檢查的左側會顯示一個英文感嘆號圖標，並且檢查下方會顯示查看錯誤詳情按鈕。點擊該按鈕後，系統會顯示一個彈出式窗口，其中說明了檢查未通過的原因以及如何糾正問題。進行適當的更正後，請點擊重新驗證。
5. 如需詳細了解如何排查未通過的驗證檢查，請參閱診斷問題。
6. 所有驗證檢查都通過後，點擊創建。
7. 在創建數據流?對話框中，點擊創建。

創建數據流後，您可以啟動它。

啟動數據流

在本教程的上一部分中，您創建了一個數據流，但並未啟動它。您現在可以進行啟動。

在本教程中，您將分別創建和啟動數據流，以防數據流創建過程造成源數據庫負載增加。如需消除該負載，您需要創建數據流但不啟動它，然後在可產生負載時啟動數據流。

通過啟動數據流，Datastream 可以將數據、架構和表從源數據庫轉移到目標位置。

1. 選擇要啟動的數據流左側的複選框。在本教程中，這是我的數據流。
2. 點擊啟動。
3. 在對話框中，點擊啟動。數據流的狀態從 Not started 更改為 Starting，再更改為 Running。
4. 數據流大約需要 30 秒才能開始運行。需要啟動後台資源，然後才能啟動數據流。

啟動數據流後，您可以驗證 Dataflow 是否將數據從源數據庫轉移到目標位置。

驗證數據流

在本部分中，您將確認 Dataflow 將數據從源 MySQL 數據庫的所有錶轉移到 Cloud Storage 目標存儲桶的 /integration/tutorial 文件夾中。在本教程中，您的存儲桶的名稱為 my-integration-bucket-001頁面。

1. 您點擊的鏈接由存儲桶的名稱以及 Datastream 將架構、表和數據從源數據庫轉移到的存儲桶的文件夾組成。在為 Cloud Storage 創建目標連接配置文件時，您已將此文件夾指定為 /integration/tutorial。因此，鏈接應顯示為 my-integration-bucket-001/integration/tutorial。
2. 驗證您是否看到表示源數據庫的表的文件夾。
3. 點擊其中一個表文件夾並展開細目，直到您看到與該表關聯的數據。
4. 第一個文件夾是 [schema]_[table]，後跟文件夾表示 Datastream 將數據從源數據庫轉移到 Cloud Storage 中的目標存儲桶的具體時間（年、月、日、小時和分鐘）。
5. 每分鐘創建一個文件夾（當要寫入新數據時）。
6. 當文件大小達到 250 MB 或架構發生更改時，系統將創建一個新文件。如果表已分區，則系統會為每個分區創建文件。

創建 Dataflow 作業

在本部分中，您將在 Dataflow 中創建作業。在 Dataflow 將源 MySQL 數據庫中的數據更改流式傳輸到 Cloud Storage 存儲桶後，系統會向 Dataflow 發送包含更改的新文件的通知。Dataflow 作業會處理這些文件並將更改轉移到 BigQuery。

1. 點擊從模板創建作業。
2. 在從模板創建作業頁面的作業名稱字段中，輸入您要創建的 Dataflow 作業的名稱。在本教程中，在字段中輸入 my-dataflow-integration-job。
3. 從地區端點菜單中，選擇存儲作業的地區。此地區就是您為所創建的來源連接配置文件、目標連接配置文件和數據流選擇的地區。
4. 從 Dataflow 模板菜單中，選擇要用於創建作業的模板。在本教程中，請選擇 Datastream to BigQuery。
5. 進行選擇後，將顯示與此模板相關的其他字段。
6. Datastream to BigQuery 模板是一種流處理流水線，可讀取 Datastream 數據並將其複製到 BigQuery 中。該模板使用 Pub/Sub 通知從 Cloud Storage 中讀取數據，並將其複製到時間分區的 BigQuery 暫存表中。複製後，該模板會在 BigQuery 中執行 MERGE，將所有變更數據捕獲獲 (CDC) 更改插入/更新到源表的副本中。
7. 為了最大限度降低頻繁 MERGE 操作產生的費用，我們建議首先採用初始頻率，即每 6-12 小時操作一次。完成所有回填且順利複製數據後，請將此值降低至所需的頻率。
8. 如需詳細了解 Datastream to BigQuery 模板，請參閱 Datastream to BigQuery (Stream)。
9. 在 Cloud Storage 中 Datastream 文件輸出的文件位置字段中，輸入包含 Cloud Storage 存儲桶的名稱的路徑。在本教程中，在字段中輸入 gs://my-integration-bucket-001。
10. 在 Cloud Storage 通知政策中使用的 Pub/Sub 訂閱字段中，輸入包含 Pub/Sub 訂閱名稱的路徑。在本教程中，在字段中輸入 projects/project-name/subscriptions/my_integration_notifs_sub。
11. project-name 是您的 Google Cloud 項目名稱的佔位符。此外，您在本教程的為 Cloud Storage 存儲桶啟用 Pub/Sub 通知部分創建了 my_integration_notifs_sub 訂閱。
12. 在 Datastream 輸出文件格式 (avro/json) 字段中輸入 avro，因為在本教程中，Avro 是 Datastream 寫入 Cloud Storage 的文件的格式。
13. 在包含暫存表的數據集的名稱或模板字段中輸入 my-dataflow-integration-job，因為 Dataflow 將使用此數據集暫存從 Datastream 接收的數據更改。
14. 在包含副本表的數據集的模板字段中輸入 My_integration_dataset_final，因為將在此數據集中合併暫存在 My_integration_dataset_log 數據集中的更改，以創建源數據庫中表的一對一副本。
15. 您在本教程的在 BigQuery 中創建數據集部分創建了 My_integration_dataset_log 和 My_integration_dataset_final 數據集。
16. 在死信隊列目錄字段中，輸入包含 Cloud Storage 存儲桶的名稱和死信隊列文件夾的路徑。Dataflow 無法轉移到 BigQuery 的數據更改都將存儲在隊列中。您可以修復隊列中的內容，以便 Dataflow 可以重新處理它。
17. 在本教程中，請在死信隊列目錄字段中輸入 gs://my-integration-bucket-001/（其中，dlq 是死信隊列的文件夾）。
18. 點擊運行作業。

您的 Dataflow 作業可能需要幾分鐘時間才能運行。需要運行後台資源。此外，作業必須先處理包含從 Cloud Storage 接收的數據更改的文件，然後才能將這些更改轉移到 BigQuery 中。

驗證集成

在本教程的驗證數據流部分中，您已確認 Datastream 將數據從源 MySQL 數據庫的所有錶轉移到 Cloud Storage 目標存儲桶的 /integration/tutorial 文件夾中。

在本部分中，您將驗證 Dataflow 是否處理包含與此數據關聯的文件，以及是否將更改傳輸到 BigQuery。因此，Datastream 和 BigQuery 之間具有端到端集成。

1. 轉到 Google Cloud Console 中 BigQuery 的 SQL 工作區頁面。
2. 在探索器窗格中，展開 Google Cloud 項目名稱左側的節點。
3. 展開 My_integration_dataset_log 和 My_integration_dataset_final 數據集左側的節點。
4. 驗證每個數據集現在是否包含數據。這會確認 Dataflow 已處理與 Datastream 流式傳輸到 Cloud Storage 中的數據相關聯的文件，並且已將這些更改轉移到 BigQuery 中。

問題描述：

問題一：在操作中需要把sql binlog 日誌開啟

問題二：在 Cloud Storage 中 Datastream 文件輸出的文件位置精確到文件avro