將資料匯出至 Pub/Sub (反向 ETL)

從 BigQuery 到 Pub/Sub 的反向擷取、轉換、載入 (RETL) 可讓您以事件導向的方式，將資料提供給下游應用程式和服務。這項工作流程結合了 BigQuery 的數據分析功能，以及 Pub/Sub 的非同步和可擴充全域訊息服務。

如要將資料匯出至 Pub/Sub，請在持續查詢中使用 EXPORT DATA 陳述式，將資料從 BigQuery 匯出至 Pub/Sub 主題。

必要條件

您必須建立服務帳戶。您必須具備服務帳戶，才能執行持續查詢，將結果匯出至 Pub/Sub 主題。

您必須建立 Pub/Sub 主題，以訊息形式接收持續查詢結果，並建立 Pub/Sub 訂閱項目，讓目標應用程式接收這些訊息。

必要的角色

本節說明建立持續查詢的使用者帳戶，以及執行持續查詢的服務帳戶，分別需要哪些角色和權限。

使用者帳戶權限

如要在 BigQuery 中建立工作，使用者帳戶必須具備 bigquery.jobs.create IAM 權限。下列各項 IAM 角色會授予 bigquery.jobs.create 權限：

如要提交使用服務帳戶執行的工作，使用者帳戶必須具備服務帳戶使用者 (roles/iam.serviceAccountUser) 角色。如果您使用相同的使用者帳戶建立服務帳戶，則使用者帳戶必須具備服務帳戶管理員 (roles/iam.serviceAccountAdmin) 角色。如要瞭解如何限制使用者只能存取單一服務帳戶，而非專案中的所有服務帳戶，請參閱「授予單一角色」。

如果使用者帳戶必須啟用持續查詢用途所需的 API，則須具備服務使用情形管理員 (roles/serviceusage.serviceUsageAdmin) 角色。

服務帳戶權限

如要從 BigQuery 資料表匯出資料，服務帳戶必須具備 bigquery.tables.export IAM 權限。下列各項 IAM 角色都會授予 bigquery.tables.export 權限：

如要讓服務帳戶存取 Pub/Sub，您必須將下列兩個 IAM 角色授予服務帳戶：

您或許也能透過自訂角色取得必要權限。

事前準備

啟用 BigQuery 和 Pub/Sub API (如果尚未啟用)。

啟用 API 所需的角色

您必須具備 serviceusage.services.enable 權限，才能啟用 API。如果您建立了專案，可能已透過「擁有者」角色 (roles/owner) 取得這項權限。否則，您可以透過「Service Usage 管理員」角色 (roles/serviceusage.serviceUsageAdmin) 取得這項權限。瞭解如何授予角色。

啟用 API

匯出至 Pub/Sub

使用 EXPORT DATA 陳述式將資料匯出至 Pub/Sub 主題：

控制台

前往 Cloud de Confiance 控制台的「BigQuery」頁面。

前往「BigQuery」
在查詢編輯器中，依序點選「編輯」>「查詢設定」。
在「Continuous Query」(持續查詢) 部分，勾選「Use continuous query mode」(使用持續查詢模式) 核取方塊。
在「Service account」(服務帳戶) 方塊中，選取您建立的服務帳戶。
按一下 [儲存]。
在查詢編輯器中輸入下列陳述式：
```
EXPORT DATA
OPTIONS (
format = 'CLOUD_PUBSUB',
uri = 'https://pubsub.googleapis.com/projects/PROJECT_ID/topics/TOPIC_ID'
) AS
(
QUERY
);
```
更改下列內容：
- PROJECT_ID：您的專案 ID。
- TOPIC_ID：Pub/Sub 主題 ID。您可以前往 Cloud de Confiance 控制台的「Topics」(主題) 頁面取得主題 ID。
- QUERY：用於選取要匯出資料的 SQL 陳述式。SQL 陳述式只能包含支援的操作。您必須在持續查詢的 FROM 子句中使用 APPENDS 函式，指定開始處理資料的時間點。
按一下「執行」。

bq

在 Cloud de Confiance 控制台啟用 Cloud Shell。

啟用 Cloud Shell

Cloud de Confiance 控制台底部會開啟 Cloud Shell 工作階段，並顯示指令列提示。Cloud Shell 是已安裝 Google Cloud CLI 的殼層環境，並已設定適用於您目前專案的值。工作階段可能要幾秒鐘的時間才能初始化。
在指令列中，使用 bq query 指令和下列旗標執行持續查詢：
- 將 --continuous 旗標設為 true，即可持續查詢。
- 使用 --connection_property 旗標指定要使用的服務帳戶。
```
bq query --project_id=PROJECT_ID --use_legacy_sql=false \
--continuous=true --connection_property=service_account=SERVICE_ACCOUNT_EMAIL \
'EXPORT DATA OPTIONS (format = "CLOUD_PUBSUB", uri = "https://pubsub.googleapis.com/projects/PROJECT_ID/topics/TOPIC_ID") AS (QUERY);'
```
更改下列內容：
- PROJECT_ID：您的專案 ID。
- SERVICE_ACCOUNT_EMAIL：服務帳戶電子郵件地址。您可以在 Cloud de Confiance 控制台的「Service accounts」(服務帳戶) 頁面取得服務帳戶電子郵件。
- QUERY：用於選取要匯出資料的 SQL 陳述式。SQL 陳述式只能包含支援的操作。您必須在持續查詢的 FROM 子句中使用 APPENDS 函式，指定開始處理資料的時間點。

API

呼叫 jobs.insert 方法，執行持續查詢。在傳入的 Job 資源中，設定 JobConfigurationQuery 資源的下列欄位：
- 將 continuous 欄位設為 true，即可持續執行查詢。
- 使用 connection_property 欄位指定要使用的服務帳戶。
```
curl --request POST \
  'https://bigquery.googleapis.com/bigquery/v2/projects/PROJECT_ID/jobs'
  --header 'Authorization: Bearer $(gcloud auth print-access-token) \
  --header 'Accept: application/json' \
  --header 'Content-Type: application/json' \
  --data '("configuration":("query":"EXPORT DATA OPTIONS (format = 'CLOUD_PUBSUB', uri = 'https://pubsub.googleapis.com/projects/PROJECT_ID/topics/TOPIC_ID') AS (QUERY);","useLegacySql":false,"continuous":true,"connectionProperties":["key": "service_account","value":"SERVICE_ACCOUNT_EMAIL"]))' \
  --compressed
```
更改下列內容：
- PROJECT_ID：您的專案 ID。
- QUERY：用於選取要匯出資料的 SQL 陳述式。SQL 陳述式只能包含支援的操作。您必須在持續查詢的 FROM 子句中使用 APPENDS 函式，指定開始處理資料的時間點。
- SERVICE_ACCOUNT_EMAIL：服務帳戶電子郵件地址。您可以在 Cloud de Confiance 控制台的「Service accounts」(服務帳戶) 頁面取得服務帳戶電子郵件。

將多個資料欄匯出至 Pub/Sub

如要在輸出內容中加入多個資料欄，可以建立 struct 資料欄來包含資料欄值，然後使用 TO_JSON_STRING 函式將 struct 值轉換為 JSON 字串。下列範例會從四個資料欄匯出資料，並格式化為 JSON 字串：

EXPORT DATA
  OPTIONS (
    format = 'CLOUD_PUBSUB',
    uri = 'https://pubsub.googleapis.com/projects/myproject/topics/taxi-real-time-rides')
AS (
  SELECT
    TO_JSON_STRING(
      STRUCT(
        ride_id,
        timestamp,
        latitude,
        longitude)) AS message
  FROM
    APPENDS(TABLE `myproject.real_time_taxi_streaming.taxi_rides`,
      -- Configure the APPENDS TVF start_timestamp to specify when you want to
      -- start processing data using your continuous query.
      -- This example starts processing at 10 minutes before the current time.
      CURRENT_TIMESTAMP() - INTERVAL 10 MINUTE)
  WHERE ride_status = 'enroute'
);

匯出最佳化

如果持續查詢工作的效能似乎受限於可用運算資源，請嘗試增加 BigQuery CONTINUOUS 運算單元預留項目指派的大小。

限制

匯出的資料必須包含單一 STRING 或 BYTES 欄。欄名稱可任意選擇。
您必須使用持續查詢，才能匯出至 Pub/Sub。
您無法在持續查詢中，將結構定義傳遞至 Pub/Sub 主題。
您無法將資料匯出至使用結構定義的 Pub/Sub 主題。
匯出至 Pub/Sub 時，您可以匯出 JSON 格式的記錄，其中部分值為 NULL，但無法匯出僅由 NULL 值組成的記錄。您可以在持續查詢中加入 WHERE message IS NOT NULL 篩選器，從查詢結果中排除 NULL 記錄。
將資料匯出至設有位置端點的 Pub/Sub 主題時，端點必須與包含查詢資料表的 BigQuery 資料集位於相同的 Cloud de Confiance 區域邊界內。
匯出的資料不得超過 Pub/Sub 配額。

定價

在持續查詢中匯出資料時，系統會依據 BigQuery 運算資源定價計費。如要執行持續查詢，您必須具備使用 Enterprise 或 Enterprise Plus 版本的預留項目，以及使用 CONTINUOUS 工作類型的預留項目指派作業。

資料匯出後，系統會向您收取 Pub/Sub 使用費。詳情請參閱「Pub/Sub 定價」。