使用 Datastream API

Datastream

對於擁有許多獨立資料來源的企業來說,要存取整個機構的企業資料 (尤其是即時存取) 可能相當困難。這會導致資料存取速度變慢,並限制資料存取量,導致組織無法進行自我檢視。

Datastream 可讓您近乎即時存取各種內部部署和雲端資料來源的變更資料,以便存取組織資料。Datastream 提供簡單的設定體驗和統一消費 API,讓機構組織能取得最新的企業資料,並支援近乎即時的整合情境。

其中一個情境是將資料從來源資料庫傳輸至雲端儲存空間服務或訊息佇列,並將這項資料轉換為可供與該儲存空間服務或訊息佇列通訊的其他應用程式和服務讀取的格式。

在本教學課程中,您將瞭解如何使用 Datastream,將架構、資料表和資料從來源 Oracle 資料庫轉移至 Cloud Storage 值區中的資料夾。Cloud Storage 是網路服務,可用於儲存及存取 Google Cloud上的資料。這項服務結合了 Google 雲端的效能和可擴充性,以及先進的安全性和共用功能。

在將這項資訊轉移至目標 Cloud Storage 值區的資料夾時,Datastream 會將這項資訊轉換為 Avro。Avro 是由以 JavaScript Object Notation (JSON) 編寫的結構定義所定義。這項轉譯功能可讓您以統一的方式讀取不同資料來源的資料。

目標

在本教學課程中,您將瞭解以下內容:

  • 設定環境變數。您向 Datastream 提出要求時,會使用這些變數來建立及管理連線設定檔串流
  • 為來源資料庫和 Cloud Storage 中的目的地值區建立及管理連線設定檔。建立這些連線設定檔後,您就會建立記錄,其中包含來源資料庫和目的地 Cloud Storage 值區的相關資訊。Datastream 中的串流會使用連線設定檔中的資訊,將資料從來源資料庫傳輸至目的地值區中的資料夾。
  • 建立及管理串流。Datastream 會使用這個串流,將資料、結構定義和資料表從來源資料庫轉移至目的地值區的資料夾。
  • 確認 Datastream 將與來源 Oracle 資料庫結構定義相關聯的資料和資料表,轉移至目的地儲存桶中的資料夾,並將這些資料轉換為 Avro 檔案格式。
  • 清除在 Datastream 上建立的資源,這樣資源就不會占用配額,您日後也無須為其付費。

費用

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

  • Cloud Storage

您可以使用 Pricing Calculator 根據預測用量產生預估費用。 新 Google Cloud 使用者可能符合申請免費試用的資格。

事前準備

  1. Sign in to your Google Cloud account. If you're new to Google Cloud, create an account to evaluate how our products perform in real-world scenarios. New customers also get $300 in free credits to run, test, and deploy workloads.
  2. In the Google Cloud console, on the project selector page, select or create a Google Cloud project.

    Go to project selector

  3. Make sure that billing is enabled for your Google Cloud project.

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

    Go to project selector

  5. Make sure that billing is enabled for your Google Cloud project.

  6. 啟用 Datastream API。

    啟用 API

  7. 請確認您已為使用者帳戶指派 Datastream 管理員角色。

    前往「身分與存取權管理」頁面

  8. 請確認您有可供 Datastream 存取的來源資料庫。在本教學課程中,我們會使用 Oracle 資料庫做為來源。
  9. 將來源資料庫設為接受來自 Datastream 公開 IP 位址的連入連線。請參閱「IP 許可清單和區域」,瞭解所有 Datastream 區域的位置,以及相關聯的公開 IP 位址。
  10. 請確認您已使用 IP 許可清單、轉送 SSH 通道或虛擬私有雲對等互連的網路連線方法,設定 Datastream 可存取的 Cloud Storage 目的地儲存桶。
  11. 請確認來源資料庫中含有資料、資料表和結構定義,以便 Datastream 將資料轉移至目的地 Cloud Storage 值區的資料夾。
  12. 下載並安裝 Cloud Shell。這個用戶端應用程式可讓您使用指令列存取雲端資源 (包括 Datastream)。
  13. 安裝及設定 jq 公用程式。這個公用程式是輕量且靈活的指令列 JSON 處理器。您將使用這個處理器,以易讀文字顯示複雜的 cURL 指令。

設定環境變數

在這個程序中,您將設定下列變數:

  • $PROJECT:此變數與您的 Google Cloud 專案相關聯。您分配及使用的任何 Google Cloud 資源都必須屬於專案。
  • $TOKEN:這個變數與存取權杖相關聯。存取權憑證會提供一個工作階段,Cloud Shell 會使用這個工作階段,透過 REST API 在 Datastream 中執行工作。
  1. 啟動 Cloud Shell 應用程式。

  2. 使用 Google 帳戶驗證應用程式後,請輸入 gcloud auth login

  3. Do you want to continue (Y/n)? 提示訊息中輸入 Y

  4. 開啟網路瀏覽器,然後將網址複製到瀏覽器中。

  5. 使用 Google 帳戶向 Google Cloud SDK 進行驗證。「登入」頁面會顯示一組代碼。這組代碼就是您的存取權杖。

  6. 複製存取權權杖,貼到 Cloud Shell 應用程式中的 Enter verification code: 參數,然後按下 Enter

  7. 在提示訊息中輸入 PROJECT="YOUR_PROJECT_NAME",將 $PROJECT 環境變數設為 Google Cloud 專案。

  8. 在提示訊息中輸入 gcloud config set project YOUR_PROJECT_NAME,即可將您要處理的專案設為 Google Cloud 專案。

    命令提示字元會更新,以反映您目前的專案,並遵循以下格式:USERNAME@cloudshell:~ (YOUR_PROJECT_NAME)$

  9. 在提示訊息中輸入 TOKEN=$(gcloud auth print-access-token),即可擷取存取權杖並儲存為變數。

  10. 在提示字元中輸入下列指令,確認 $PROJECT$TOKEN 變數設定正確無誤:

    • echo $PROJECT
    • echo $TOKEN

設定變數後,您可以向 Datastream 提出要求,建立及管理連線設定檔和串流。

建立及管理連線設定檔

在本節中,您將為來源 Oracle 資料庫和 Cloud Storage 中的目的地值區建立及管理連線設定檔。

建立這些連線設定檔時,您會建立記錄,其中包含來源資料庫和目的地 Cloud Storage 值區的相關資訊。Datastream 會使用連線設定檔中的資訊,將資料從來源資料庫傳輸至目的地值區中的資料夾。

建立及管理連線設定檔的作業包括:

  • 為 Oracle 來源資料庫和 Cloud Storage 中的目的地值區建立連線設定檔
  • 擷取連線設定檔的相關資訊
  • 修改連線設定檔
  • 對來源 Oracle 連線設定檔執行 discover API 呼叫。這個呼叫可讓您查看資料庫內部,查看與資料庫相關聯的物件。這些物件包括包含資料庫資料的結構定義和資料表。使用 Datastream 設定串流時,您可能不想從資料庫中提取所有物件,而是只提取物件的子集 (例如,只提取資料庫的特定資料表和結構定義)。使用 discover API 找出 (或探索) 您要擷取的資料庫物件子集。

建立連線設定檔

在這個程序中,您會建立兩個連線設定檔:一個連線至來源 Oracle 資料庫,另一個連線至 Cloud Storage 中的目的地值區。

  1. 建立連線設定檔,連線至來源 Oracle 資料庫。在提示中輸入下列指令:
ORACLE="{\"displayName\":\"DISPLAY_NAME\",\"oracle_profile\":{\"hostname\":\"HOSTNAME\",\"username\":\"USERNAME\",\"password\":\"PASSWORD\",\"database_service\":\"DATABASE_SERVICE\",\"port\":"PORT_NUMBER\"},\"no_connectivity\":{}}"
  

請參閱下表,瞭解來源 Oracle 資料庫的參數值:

參數值取代憑證
DISPLAY_NAME來源資料庫連線設定檔的顯示名稱。
HOSTNAME來源資料庫伺服器的主機名稱。
USERNAME來源資料庫帳戶的使用者名稱 (例如 ROOT)。
PASSWORD來源資料庫帳戶的密碼。
DATABASE_SERVICE這項服務可確保來源資料庫受到保護及監控。對於 Oracle 資料庫,資料庫服務通常為 ORCL
PORT_NUMBER來源資料庫保留的通訊埠號碼。對於 Oracle 資料庫,通訊埠編號通常為 1521。

  1. 在提示中輸入 echo $ORACLE | jq 指令,即可透過易讀的文字查看您建立的來源連線設定檔。

    {
      "displayName": "DISPLAY_NAME",
      "oracle_profile": {
        "hostname": "HOSTNAME",
        "username": "USERNAME",
        "password": "PASSWORD",
        "database_service": "DATABASE_SERVICE",
        "port": PORT_NUMBER
       },
      "no_connectivity": {}
    }
  2. 提交 Oracle 連線設定檔,以便建立連線設定檔。在提示中輸入下列指令:

    curl -X POST -d $ORACLE -H "Authorization: Bearer $TOKEN" -H "Content-Type: application/json" https://datastream.googleapis.com/DATASTREAM_API_VERSION/PROJECT_PATH/connectionProfiles?connection_profile_id=SOURCE_CONNECTION_PROFILE_ID

    請參閱下表,瞭解這個指令的參數值:

    參數值取代憑證
    DATASTREAM_API_VERSIONDatastream API 的目前版本 (例如 v1)。
    PROJECT_PATH Google Cloud 專案的完整路徑 (例如 projects/$PROJECT/locations/YOUR_PROJECT_LOCATION)。
    SOURCE_CONNECTION_PROFILE_ID此連線設定檔的專屬 ID (例如 cp-1)。
  3. 確認您看到以下程式碼行:

    {
      "name": "PROJECT_PATH/operations/operation-SOURCE_CONNECTION_PROFILE_OPERATION_ID",
      "metadata": {
        "@type": "type.googleapis.com/google.cloud.datastream.DATASTREAM_API_VERSION.OperationMetadata",
        "createTime": "DATE_AND_TIME_STAMP",
        "target": "datastream.googleapis.com/DATASREAM_VERSION/PROJECT_PATH/connectionProfiles/SOURCE_CONNECTION_PROFILE_ID",
        "verb": "create",
        "requestedCancellation": false,
        "apiVersion": "DATASTREAM_API_VERSION"
      },
      "done": false
    }
  4. 建立 Cloud Storage 中目的地值區的連線設定檔。在提示中輸入下列指令:

    GOOGLECLOUDSTORAGE="{\"displayName\":\"DISPLAY_NAME\",\"gcs_profile\":{\"bucket_name\":\"BUCKET_NAME\",\"root_path\":\"/FOLDER_PATH\"},\"no_connectivity\":{}}"

    請參閱下表,瞭解目的地值區的參數值:

    參數值取代憑證
    DISPLAY_NAME目的地資料集連線設定檔的顯示名稱。
    BUCKET_NAME目標值區名稱。
    FOLDER_PATH目的地值區中的資料夾,Datastream 會將來源資料庫 (例如 /root/path) 中的資料轉移至此資料夾。
  5. 在提示中輸入 echo $GOOGLECLOUDSTORAGE | jq 指令,即可透過易讀的文字查看您建立的目的地連線設定檔。

    {
      "displayName": "DISPLAY_NAME",
      "gcs_profile": {
        "bucket_name": "BUCKET_NAME",
        "root_path": "/FOLDER_PATH"
      },
      "no_connectivity": {}
    }
  6. 提交 Cloud Storage 連線設定檔,即可建立連線。在提示中輸入下列指令:

    curl -X POST -d $GOOGLECLOUDSTORAGE -H "Authorization: Bearer $TOKEN" -H "Content-Type: application/json" https://datastream.googleapis.com/DATASTREAM_API_VERSION/PROJECT_PATH/connectionProfiles?connection_profile_id=DESTINATION_CONNECTION_PROFILE_ID
  7. 確認您看到以下程式碼行:

    {
      "name": "PROJECT_PATH/operations/operation-DESTINATION_CONNECTION_PROFILE_OPERATION_ID",
      "metadata": {
        "@type": "@type": "type.googleapis.com/google.cloud.datastream.DATASTREAM_API_VERSION.OperationMetadata",
        "createTime": "DATE_AND_TIME_STAMP",
        "target": "datastream.googleapis.com/DATASTREAM_VERSION/PROJECT_PATH/connectionProfiles/DESTINATION_CONNECTION_PROFILE_ID",
        "verb": "create",
        "requestedCancellation": false,
        "apiVersion": "DATASTREAM_API_VERSION"
      },
      "done": false
    }
  8. 確認已建立這兩個連線設定檔。在提示中輸入下列指令:

    curl -H "Authorization: Bearer $TOKEN" -H "Content-Type: application/json" https://datastream.googleapis.com/DATASTREAM_API_VERSION/PROJECT_PATH/connectionProfiles
  9. 確認您已收到來源和目的地連線設定檔的兩個傳回結果。

    {
      "connectionProfiles": [
        {
          "name": "PROJECT_PATH/connectionProfiles/DESTINATION_CONNECTION_PROFILE_ID",
          "createTime": "DATE_AND_TIME_STAMP",
          "updateTime": "DATE_AND_TIME_STAMP",
          "displayName": "DISPLAY_NAME",
          "gcsProfile": {
            "bucketName": "BUCKET_NAME",
            "rootPath": "FOLDER_PATH"
          },
          "noConnectivity": {}
        },
       {
        "name": "PROJECT_PATH/connectionProfiles/SOURCE_CONNECTION_PROFILE_ID",
        "createTime": "DATE_AND_TIME_STAMP",
        "updateTime": "DATE_AND_TIME_STAMP",
        "displayName": "DISPLAY_NAME",
        "oracleProfile": {
          "hostname": "HOSTNAME",
          "port": PORT_NUMBER,
          "username": "USERNAME",
          "databaseService": "DATABASE_SERVICE"
        },
        "noConnectivity": {}
        }
      ]
    }

管理連線設定檔

在這個程序中,您將管理為來源 Oracle 資料庫和 Cloud Storage 中的目的地值區建立的連線設定檔。包括:

  • 擷取目的地 Cloud Storage 連線設定檔的相關資訊
  • 修改這個連線設定檔。在本教學課程中,您將將目的地 Cloud Storage 值區的資料夾變更為 /root/tutorial。Datastream 會將資料從來源資料庫轉移至這個資料夾。
  • 對來源 Oracle 連線設定檔執行 discover API 呼叫
  1. 擷取目的地 Cloud Storage 連線設定檔的相關資訊。在提示中輸入下列指令:

    curl -H "Authorization: Bearer $TOKEN" -H "Content-Type: application/json" https://datastream.googleapis.com/DATASTREAM_API_VERSION/PROJECT_PATH/connectionProfiles/DESTINATION_CONNECTION_PROFILE_ID
  2. 確認您能看到這個連線設定檔的相關資訊。

    {
      "name": "PROJECT_PATH/connectionProfiles/DESTINATION_CONNECTION_PROFILE_ID",
      "createTime": "DATE_AND_TIME_STAMP",
      "updateTime": "DATE_AND_TIME_STAMP",
      "displayName": "DISPLAY_NAME",
      "gcsProfile": {
        "bucketName": "BUCKET_NAME",
        "rootPath": "FOLDER_PATH"
      },
      "noConnectivity": {}
    }
  3. 修改這個連線設定檔。如要這麼做,請先設定 UPDATE 變數。這個變數包含您要變更的連線設定檔值。在本教學課程中,您會將目的地值區的資料夾變更為 /root/tutorial

    如要設定變數,請在提示字元中輸入下列指令:

    UPDATE="{\"gcsProfile\":{\"rootPath\":\"/root/tutorial\"}}"
  4. 在提示中輸入下列指令:

    curl -X PATCH -d $UPDATE -H "Authorization: Bearer $TOKEN" -H "Content-Type: application/json" https://datastream.googleapis.com/DATASTREAM_API_VERSION/PROJECT_PATH/connectionProfiles/DESTINATION_CONNECTION_PROFILE_ID?update_mask=gcsProfile.rootPath
  5. 確認您看到以下程式碼行:

    {
      "name": "PROJECT_PATH/operations/operation-DESTINATION_CONNECTION_PROFILE_OPERATION_ID",
      "metadata": {
        "@type": "type.googleapis.com/google.cloud.datastream.DATASTREAM_API_VERSION.OperationMetadata",
        "createTime": "DATE_AND_TIME_STAMP",
        "target": "PROJECT_PATH/connectionProfiles/DESTINATION_CONNECTION_PROFILE_ID",
        "verb": "update",
        "requestedCancellation": false,
        "apiVersion": "DATASTREAM_API_VERSION"
      },
      "done": false
    }
  6. 確認已修改連線設定檔。在提示中輸入下列指令:

    curl -H "Authorization: Bearer $TOKEN" -H "Content-Type: application/json" https://datastream.googleapis.com/DATASTREAM_API_VERSION/PROJECT_PATH/connectionProfiles/DESTINATION_CONNECTION_PROFILE_ID
  7. 確認 Cloud Storage 連線設定檔的目的地值區資料夾現在是 /root/tutorial

    {
      "name": "PROJECT_PATH/connectionProfiles/DESTINATION_CONNECTION_PROFILE_ID",
      "createTime": "DATE_AND_TIME_STAMP",
      "updateTime": "DATE_AND_TIME_STAMP",
      "displayName": "DISPLAY_NAME",
      "gcsProfile": {
        "bucketName": "BUCKET_NAME",
        "rootPath": "/root/tutorial"
      },
      "noConnectivity": {}
    }
  8. 使用 Datastream 探索 API 探索來源 Oracle 資料庫的結構定義和資料表。Datastream 會透過來源連線設定檔提供存取權。

    1. 探索 Oracle 資料庫的結構定義。在提示中輸入下列指令:

      curl -X POST -d "{\"connection_profile_name\":\"projects/YOUR_PROJECT_NUMBER/locations/YOUR_PROJECT_LOCATION/connectionProfiles/SOURCE_CONNECTION_PROFILE_ID\", \"oracle_rdbms\":{\"oracleSchemas\":[{\"schema\":\"ROOT\"}]}}" -H "Authorization: Bearer $TOKEN" -H "Content-Type: application/json" https://datastream.googleapis.com/DATASTREAM_API_VERSION/PROJECT_PATH/connectionProfiles:discover
    2. 確認 Datastream 擷取資料庫的所有結構定義。

    3. 擷取資料庫中結構定義的資料表。在本教學課程中,您將使用 discover API 擷取 ROOT 結構定義的資料表。不過,您可以探索資料庫中任何結構定義的資料表。

      在提示中輸入下列指令:

    curl -X POST -d "{\"connection_profile_name\":\"projects/YOUR_PROJECT_NUMBER/locations/YOUR_PROJECT_LOCATION/connectionProfiles/SOURCE_CONNECTION_PROFILE_ID\", \"oracle_rdbms\":{\"oracleSchemas\":[{\"schema\":\"ROOT\"}]}}" -H "Authorization: Bearer $TOKEN" -H "Content-Type: application/json" https://datastream.googleapis.com/DATASTREAM_API_VERSION/PROJECT_PATH/connectionProfiles:discover
    1. 確認 Datastream 擷取您指定結構定義的所有資料表 (本教學課程為 ROOT 結構定義)。

您已為 Oracle 來源資料庫和 Cloud Storage 中的目的地值區建立及管理連線設定檔,現在可以開始在 Datastream 中建立及管理串流了。

建立及管理串流

在本節中,您將建立及管理串流。Datastream 會使用這個串流,將資料、結構定義和資料表從來源資料庫傳輸至目的地 Cloud Storage 值區的資料夾。

建立及管理串流的作業包括:

  • 驗證串流,確保串流能順利執行,且所有驗證檢查都通過。這些檢查項目包括:
    • 來源是否已正確設定,讓 Datastream 能夠從中串流資料。
    • 串流是否可以連線至來源和目的地。
    • 串流的端對端設定。
  • 使用下列清單建立串流:
    • 許可清單。這份清單會指定 Datastream 可將來源資料庫中的資料表和結構定義轉移至 Cloud Storage 目標值區的資料夾。在本教學課程中,這個資料夾為 /root/tutorial
    • 拒絕清單。這份清單會指定來源資料庫中的資料表和結構定義,這些項目會受到限制,無法由 Datastream 轉移至 Cloud Storage 目的地值區的資料夾。
  • 擷取串流相關資訊
  • 修改串流
  • 啟動串流,讓 Datastream 將資料、結構定義和資料表從來源資料庫傳輸至目的地 Cloud Storage 值區的資料夾。
  • 使用 Fetch Errors API 偵測與串流相關的任何錯誤
  • 暫停串流。暫停串流後,Datastream 就不會從來源資料庫將任何新資料擷取到目標儲存格中。
  • 繼續暫停的串流,讓 Datastream 繼續將資料移至目的地值區。

建立串流

在這個程序中,您會建立從來源 Oracle 資料庫串流至目的地 Cloud Storage 值區資料夾的串流。您建立的串流會包含允許清單和拒絕清單。

  1. 設定 SCHEMAS 變數。這個變數會定義結構定義,其中包含您希望 Datastream 從來源資料庫擷取的資料和資料表,並轉移至 Cloud Storage 目的地值區的 /root/tutorial 資料夾。在本教學課程中,您會將 SCHEMAS 變數設為與 ROOT 結構定義相關聯。

    在提示中輸入下列指令:

    SCHEMAS="{\"oracleSchemas\":[{\"schema\":\"ROOT\"}]}"
  2. 在提示訊息中輸入 echo $SCHEMAS | jq 指令,即可查看您為這個變數定義的 ROOT 結構定義,並以易讀的文字呈現。

  3. 建立串流。在提示中輸入下列指令:

    STREAM="{\"display_name\":\"DISPLAY_NAME\",\"source_config\":{\"source_connection_profile_name\":\"PROJECT_PATH/connectionProfiles/SOURCE_CONNECTION_PROFILE_ID",\"oracle_source_config\":{\"allowlist\":$SCHEMAS,\"rejectlist\":{}}},\"destination_config\":{\"destination_connection_profile_name\":\"PROJECT_PATH/connectionProfiles/DESTINATION_CONNECTION_PROFILE_ID\",\"gcs_destination_config\":{\"file_rotation_mb\":5,\"file_rotation_interval\":{\"seconds\":15},\"avro_file_format\":{}},\"backfill_all\":{}}}"
  4. 在提示訊息中輸入 echo $STREAM | jq 指令,即可透過易讀的文字查看您建立的串流。

    {
      "display_name": "DISPLAY_NAME",
      "source_config": {
        "source_connection_profile_name": "PROJECT_PATH/connectionProfiles/SOURCE_CONNECTION_PROFILE_ID",
        "oracle_source_config": {
          "allowlist": {
            "oracleSchemas": [
              {
                "schema": "ROOT"
              }
            ]
          },
          "rejectlist": {}
        }
      },
      "destination_config": {
        "destination_connection_profile_name": "PROJECT_PATH/connectionProfiles/DESTINATION_CONNECTION_PROFILE_ID",
        "gcs_destination_config": {
          "file_rotation_mb": 5,
          "file_rotation_interval": {
            "seconds": 15
          },
          "avro_file_format": {}
        }
      },
      "backfill_all": {}
    }

    請參閱下表,瞭解串流的下列參數:

    參數說明
    allowlist包含資料表和資料的結構定義,會從來源資料庫傳輸至 Cloud Storage 目的地值區的資料夾。在本教學課程中,來自 ROOT 結構定義的所有資料表和資料 (僅限這個結構定義) 都會轉移至目的地儲存桶的 /root/tutorial 資料夾。
    rejectlist任何包含資料表和資料的架構,都不會轉移至 Cloud Storage 目的地值區的資料夾。在本教學課程中,{} 值代表不會阻止從來源資料庫傳輸至目的地 bucket 的任何資料表和資料。
    file_rotation_mb從來源資料庫轉移至 Cloud Storage 目的地值區的資料夾中,含有資料的檔案大小 (以 MB 為單位)。在本教學課程中,系統會從來源資料庫擷取資料,並將資料寫入 5 MB 的檔案。如果任何資料超過這個大小,系統會將資料分割成多個 5 MB 的檔案。
    file_rotation_interval在 Datastream 關閉 Cloud Storage 目的地值區資料夾中現有檔案,並開啟另一個檔案來儲存從來源資料庫傳輸的資料之前,要經過的秒數。在本教學課程中,檔案輪替間隔設為 15 秒。
    avro_file_format

    Datastream 會將檔案從來源資料庫傳輸至 Cloud Storage 目的地值區的資料夾,而這裡指定的格式就是這些檔案的格式。在本教學課程中,檔案格式為 Avro。

    backfill_all

    這個參數與歷來資料填補相關。將這個參數設為空字典 ({}) 後,Datastream 會回填:

    • 從來源資料庫到目的地,除了持續變更的資料外,也包含歷來資料。
    • 從來源到目的地的結構定義和資料表。
  5. 驗證串流,確保串流能順利執行,且所有驗證檢查均通過。在提示中輸入下列指令:

    curl -X POST -d $STREAM -H "Authorization: Bearer $TOKEN" -H "Content-Type: application/json" "https://datastream.googleapis.com/DATASTREAM_API_VERSION/PROJECT_PATH/streams?stream_id=STREAM_ID&validate_only=true"
  6. 確認您看到 {} 程式碼行。這表示串流已通過所有驗證檢查,且與串流相關的錯誤皆已修正。

  7. 提交串流以便建立。在提示中輸入下列指令:

    curl -X POST -d $STREAM -H "Authorization: Bearer $TOKEN" -H "Content-Type: application/json" https://datastream.googleapis.com/DATASTREAM_API_VERSION/PROJECT_PATH/streams?stream_id=STREAM_ID
  8. 確認您看到以下程式碼行:

    {
      "name": "PROJECT_PATH/operations/operation-STREAM_OPERATION_ID",
      "metadata": {
        "@type": "type.googleapis.com/google.cloud.datastream.DATASTREAM_API_VERSION.OperationMetadata",
        "createTime": "DATE_AND_TIME_STAMP",
        "target": "PROJECT_PATH/streams/STREAM_ID",
        "verb": "create",
        "requestedCancellation": false,
        "apiVersion": "DATASTREAM_API_VERSION"
      },
      "done": false
    }
  9. 確認串流已建立。在提示中輸入下列指令:

    curl -H "Authorization: Bearer $TOKEN" -H "Content-Type: application/json" https://datastream.googleapis.com/DATASTREAM_API_VERSION/PROJECT_PATH/streams
  10. 確認您已收到所建立串流的傳回結果。

    {
      "streams": [
        {
          "name": "PROJECT_PATH/streams/STREAM_ID",
          "createTime": "DATE_AND_TIME_STAMP",
          "updateTime": "DATE_AND_TIME_STAMP",
          "displayName": "DISPLAY_NAME",
          "sourceConfig": {
            "sourceConnectionProfileName": "projects/YOUR_PROJECT_NUMBER/locations/YOUR_PROJECT_LOCATION/connectionProfiles/SOURCE_CONNECTION_PROFILE_ID",
            "oracleSourceConfig": {
              "allowlist": {
                "oracleSchemas": [
                  {
                    "schema": "ROOT"
                  }
                ]
              },
              "rejectlist": {}
            }
          },
          "destinationConfig": {
            "destinationConnectionProfileName": "projects/YOUR_PROJECT_NUMBER/locations/YOUR_PROJECT_LOCATION/connectionProfiles/DESTINATION_CONNECTION_PROFILE_ID",
          "gcsDestinationConfig": {
              "fileRotationMb": 5,
              "fileRotationInterval": "15s"
              "avroFileFormat": {}
            }
          },
          "state": "CREATED",
          "backfillAll": {}
        }
      ]
    }

管理串流

在這個程序中,您會使用建立的串流,將資料從來源 Oracle 資料庫轉移至 Cloud Storage 目的地值區中的資料夾。包括:

  • 擷取串流相關資訊
  • 修改串流
  • 開始串流
  • 使用 Fetch Errors API 偵測與串流相關的任何錯誤
  • 暫停及繼續串流
  1. 擷取串流的相關資訊。在提示中輸入下列指令:

    curl -H "Authorization: Bearer $TOKEN" -H "Content-Type: application/json" https://datastream.googleapis.com/DATASTREAM_API_VERSION/PROJECT_PATH/streams/STREAM_ID
  2. 確認您能看到這個串流的相關資訊。

    {
      "name": "PROJECT_PATH/streams/STREAM_ID",
      "createTime": "DATE_AND_TIME_STAMP",
      "updateTime": "DATE_AND_TIME_STAMP",
      "displayName": "DISPLAY_NAME",
      "sourceConfig": {
        "sourceConnectionProfileName": "projects/YOUR_PROJECT_NUMBER/locations/YOUR_PROJECT_LOCATION/connectionProfiles/SOURCE_CONNECTION_PROFILE_ID",
        "oracleSourceConfig": {
          "allowlist": {
            "oracleSchemas": [
              {
                "schema": "ROOT"
              }
            ]
          },
          "rejectlist": {}
         }
        },
        "destinationConfig": {
          "destinationConnectionProfileName": "projects/YOUR_PROJECT_NUMBER/locations/YOUR_PROJECT_LOCATION/connectionProfiles/DESTINATION_CONNECTION_PROFILE_ID",
          "gcsDestinationConfig": {
            "fileRotationMb": 5,
            "fileRotationInterval": "15s"
            "avroFileFormat": {}
          }
        },
        "state": "CREATED",
        "backfillAll": {}
      }
  3. 修改這個串流。如要這麼做,請先設定 UPDATE 變數。這個變數包含要變更的串流值。在本教學課程中,請將從來源資料庫傳輸至 Cloud Storage 目的地值區資料夾的檔案大小 (以 MB 為單位) 從 5 MB 變更為 100 MB。從來源資料庫擷取資料時,現在會將資料寫入 100 MB 的檔案。如果任何資料超過這個大小,系統就會將資料分割成多個 100 MB 的檔案。

    如要設定變數,請在提示字元中輸入下列指令:

    UPDATE="{\"destination_config\":{\"gcs_destination_config\":{\"file_rotation_mb\":100}}}"
  4. 在提示中輸入下列指令:

    curl -X PATCH -d $UPDATE -H "Authorization: Bearer $TOKEN" -H "Content-Type: application/json" https://datastream.googleapis.com/DATASTREAM_API_VERSION/PROJECT_PATH/streams/STREAM_ID/?update_mask=destination_config.gcs_destination_config.file_rotation_mb
  5. 確認您看到以下程式碼行:

    {
      "name": "PROJECT_PATH/operations/operation-STREAM_OPERATION_ID",
      "metadata": {
        "@type": "type.googleapis.com/google.cloud.datastream.DATASTREAM_API_VERSION.OperationMetadata",
        "createTime": "DATE_AND_TIME_STAMP",
        "target": "PROJECT_PATH/streams/STREAM_ID",
        "verb": "update",
        "requestedCancellation": false,
        "apiVersion": "DATASTREAM_API_VERSION"
      },
      "done": false
    }
  6. 確認已修改串流。在提示中輸入下列指令:

    curl -H "Authorization: Bearer $TOKEN" -H "Content-Type: application/json" https://datastream.googleapis.com/DATASTREAM_API_VERSION/PROJECT_PATH/streams/STREAM_ID
  7. 確認 Cloud Storage 連線設定檔的 fileRotationMb 參數值現在為 100

    {
      "name": "PROJECT_PATH/streams/STREAM_ID",
      "createTime": "DATE_AND_TIME_STAMP",
      "updateTime": "DATE_AND_TIME_STAMP",
      "displayName": "DISPLAY_NAME",
      "sourceConfig": {
        "sourceConnectionProfileName": "projects/YOUR_PROJECT_NUMBER/locations/YOUR_PROJECT_LOCATION/connectionProfiles/SOURCE_CONNECTION_PROFILE_ID",
        "oracleSourceConfig": {
          "allowlist": {
            "oracleSchemas": [
              {
                "schema": "ROOT"
              }
            ]
          },
          "rejectlist": {}
         }
        },
        "destinationConfig": {
          "destinationConnectionProfileName": "projects/YOUR_PROJECT_NUMBER/locations/YOUR_PROJECT_LOCATION/connectionProfiles/DESTINATION_CONNECTION_PROFILE_ID",
          "gcsDestinationConfig": {
            "fileRotationMb": 100,
            "fileRotationInterval": "15s"
            "avroFileFormat": {}
          }
        },
        "state": "CREATED",
        "backfillAll": {}
      }
  8. 開始串流。方法如下:

    1. 變更 UPDATE 變數。在提示中輸入下列指令:

      UPDATE="{\"state\":\"RUNNING\"}"
    2. 然後輸入下列指令:

      curl -H "Authorization: Bearer $TOKEN" -H "Content-Type: application/json" https://datastream.googleapis.com/DATASTREAM_API_VERSION/PROJECT_PATH/streams/STREAM_ID?updateMask=state
  9. 確認您看到以下程式碼行。

    {
      "name": "PROJECT_PATH/operations/operation-STREAM_OPERATION_ID",
      "metadata": {
        "@type": "type.googleapis.com/google.cloud.datastream.DATASTREAM_API_VERSION.OperationMetadata",
        "createTime": "DATE_AND_TIME_STAMP",
        "target": "PROJECT_PATH/streams/STREAM_ID",
        "verb": "start",
        "requestedCancellation": false,
        "apiVersion": "DATASTREAM_API_VERSION"
      },
      "done": false
    }
  10. 幾分鐘後,請擷取串流的相關資訊,確認串流已開始:

    curl -H "Authorization: Bearer $TOKEN" -H "Content-Type: application/json" https://datastream.googleapis.com/DATASTREAM_API_VERSION/PROJECT_PATH/streams/STREAM_ID
  11. 確認串流的狀態已從 CREATED 變更為 RUNNING

    {
      "name": "PROJECT_PATH/streams/STREAM_ID",
      "createTime": "DATE_AND_TIME_STAMP",
      "updateTime": "DATE_AND_TIME_STAMP",
      "displayName": "DISPLAY_NAME",
      "sourceConfig": {
        "sourceConnectionProfileName": "projects/YOUR_PROJECT_NUMBER/locations/YOUR_PROJECT_LOCATION/connectionProfiles/SOURCE_CONNECTION_PROFILE_ID",
        "oracleSourceConfig": {
          "allowlist": {
            "oracleSchemas": [
              {
                "schema": "ROOT"
              }
            ]
          },
          "rejectlist": {}
        }
      },
      "destinationConfig": {
        "destinationConnectionProfileName": "projects/YOUR_PROJECT_NUMBER/locations/YOUR_PROJECT_LOCATION/connectionProfiles/DESTINATION_CONNECTION_PROFILE_ID",
        "gcsDestinationConfig": {
          "fileRotationMb": 100,
          "fileRotationInterval": "15s"
          "avroFileFormat": {}
        }
      },
      "state": "RUNNING",
      "backfillAll": {}
    }
  12. 使用 Fetch Errors API 擷取與串流相關的任何錯誤。

    1. 在提示中輸入下列指令:

      curl -X POST -H "Authorization: Bearer $TOKEN" -H "Content-Type: application/json" https://datastream.googleapis.com/DATASTREAM_API_VERSION/PROJECT_PATH/streams/STREAM_ID:fetchErrors
    2. 確認您看到以下程式碼行:

        {
          "name": "PROJECT_PATH/operations/operation-FETCH_ERRORS_OPERATION_ID",
          "metadata": {
            "@type": "type.googleapis.com/google.cloud.datastream.DATASTREAM_API_VERSION.OperationMetadata",
            "createTime": "DATE_AND_TIME_STAMP",
            "target": "PROJECT_PATH/streams/STREAM_ID",
            "verb": "fetchErrors",
            "requestedCancellation": false,
            "apiVersion": "DATASTREAM_API_VERSION"
          },
          "done": false
        }
        

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

      curl -H "Authorization: Bearer $TOKEN" -H "Content-Type: application/json" https://datastream.googleapis.com/DATASTREAM_API_VERSION/PROJECT_PATH/operations/operation-FETCH_ERRORS_OPERATION_ID
    4. 確認您看到以下程式碼行:

        {
          "name": "PROJECT_PATH/operations/operation-FETCH_ERRORS_OPERATION_ID",
          "metadata": {
            "@type": "type.googleapis.com/google.cloud.datastream.DATASTREAM_API_VERSION.OperationMetadata",
            "createTime": "DATE_AND_TIME_STAMP",
            "endTime": "DATE_AND_TIME_STAMP",
            "target": "PROJECT_PATH/streams/STREAM_ID",
            "verb": "fetchErrors",
            "requestedCancellation": false,
            "apiVersion": "DATASTREAM_API_VERSION"
          },
          "done": true,
          "response": {
            "@type": "type.googleapis.com/google.cloud.datastream.DATASTREAM_API_VERSION.FetchErrorsResponse"
          }
        }
        

  13. 暫停串流。方法如下:

    1. 變更 UPDATE 變數。在提示中輸入下列指令:

      UPDATE="{\"state\":\"PAUSED\"}"
    2. 然後輸入下列指令:

      curl -H "Authorization: Bearer $TOKEN" -H "Content-Type: application/json" https://datastream.googleapis.com/DATASTREAM_API_VERSION/PROJECT_PATH/streams/STREAM_ID?updateMask=state
  14. 確認您看到以下程式碼行。

    {
      "name": "PROJECT_PATH/operations/operation-STREAM_OPERATION_ID",
      "metadata": {
        "@type": "type.googleapis.com/google.cloud.datastream.DATASTREAM_API_VERSION.OperationMetadata",
        "createTime": "DATE_AND_TIME_STAMP",
        "target": "PROJECT_PATH/streams/STREAM_ID",
        "verb": "start",
        "requestedCancellation": false,
        "apiVersion": "DATASTREAM_API_VERSION"
      },
      "done": false
    }
  15. 擷取串流資訊,確認串流已暫停。

    curl -H "Authorization: Bearer $TOKEN" -H "Content-Type: application/json" https://datastream.googleapis.com/DATASTREAM_API_VERSION/PROJECT_PATH/streams/STREAM_ID
  16. 確認串流的狀態已從 RUNNING 變更為 PAUSED

    {
      "name": "PROJECT_PATH/streams/STREAM_ID",
      "createTime": "DATE_AND_TIME_STAMP",
      "updateTime": "DATE_AND_TIME_STAMP",
      "displayName": "DISPLAY_NAME",
      "sourceConfig": {
        "sourceConnectionProfileName": "projects/YOUR_PROJECT_NUMBER/locations/YOUR_PROJECT_LOCATION/connectionProfiles/SOURCE_CONNECTION_PROFILE_ID",
        "oracleSourceConfig": {
          "allowlist": {
            "oracleSchemas": [
              {
                "schema": "ROOT"
              }
            ]
          },
          "rejectlist": {}
        }
      },
      "destinationConfig": {
        "destinationConnectionProfileName": "projects/YOUR_PROJECT_NUMBER/locations/YOUR_PROJECT_LOCATION/connectionProfiles/DESTINATION_CONNECTION_PROFILE_ID",
        "gcsDestinationConfig": {
          "fileRotationMb": 100,
          "fileRotationInterval": "15s"
          "avroFileFormat": {}
        }
      },
      "state": "PAUSED",
      "backfillAll": {}
    }
  17. 繼續播放已暫停的串流。方法如下:

    1. 變更 UPDATE 變數。在提示中輸入下列指令:

      UPDATE="{\"state\":\"RUNNING\"}"
    2. 然後輸入下列指令:

      curl -H "Authorization: Bearer $TOKEN" -H "Content-Type: application/json" https://datastream.googleapis.com/DATASTREAM_API_VERSION/PROJECT_PATH/streams/STREAM_ID?updateMask=state
  18. 確認您看到以下程式碼行。

    {
      "name": "PROJECT_PATH/operations/operation-STREAM_OPERATION_ID",
      "metadata": {
        "@type": "type.googleapis.com/google.cloud.datastream.DATASTREAM_API_VERSION.OperationMetadata",
        "createTime": "DATE_AND_TIME_STAMP",
        "target": "PROJECT_PATH/streams/STREAM_ID",
        "verb": "start",
        "requestedCancellation": false,
        "apiVersion": "DATASTREAM_API_VERSION"
      },
      "done": false
    }
  19. 幾秒後,請擷取串流相關資訊,確認串流已重新執行。

    curl -H "Authorization: Bearer $TOKEN" -H "Content-Type: application/json" https://datastream.googleapis.com/DATASTREAM_API_VERSION/PROJECT_PATH/streams/STREAM_ID
  20. 確認串流的狀態已從 PAUSED 變更回 RUNNING

    {
      "name": "PROJECT_PATH/streams/STREAM_ID",
      "createTime": "DATE_AND_TIME_STAMP",
      "updateTime": "DATE_AND_TIME_STAMP",
      "displayName": "DISPLAY_NAME",
      "sourceConfig": {
        "sourceConnectionProfileName": "projects/YOUR_PROJECT_NUMBER/locations/YOUR_PROJECT_LOCATION/connectionProfiles/SOURCE_CONNECTION_PROFILE_ID",
        "oracleSourceConfig": {
          "allowlist": {
            "oracleSchemas": [
              {
                "schema": "ROOT"
              }
            ]
          },
          "rejectlist": {}
        }
      },
      "destinationConfig": {
        "destinationConnectionProfileName": "projects/YOUR_PROJECT_NUMBER/locations/YOUR_PROJECT_LOCATION/connectionProfiles/DESTINATION_CONNECTION_PROFILE_ID",
        "gcsDestinationConfig": {
          "fileRotationMb": 100,
          "fileRotationInterval": "15s"
          "avroFileFormat": {}
        }
      },
      "state": "RUNNING",
      "backfillAll": {}
    }

您已建立及管理串流、確認串流沒有任何錯誤,且串流狀態為 RUNNING,現在可以驗證串流是否能將資料從來源資料庫傳輸至 Cloud Storage 目的地儲存桶中的資料夾。

驗證串流

在這個程序中,您會確認 Datastream 是否符合下列條件:

  • 將與來源 Oracle 資料庫的 ROOT 架構相關聯的所有資料表,轉移至 Cloud Storage 目的地值區的 /root/tutorial 資料夾。
  • 將資料轉換為 Avro 檔案格式。
  1. 前往 Cloud Storage 中的「Storage browser」(儲存空間瀏覽器) 頁面。

    前往「Storage Browser」(儲存空間瀏覽器) 頁面

  2. 按一下包含值區的連結。

  3. 如果「物件」分頁未啟用,請按一下該分頁。

  4. 依序點選「root」資料夾和「tutorial」資料夾。

  5. 確認您看見代表來源 Oracle 資料庫 ROOT 結構定義的資料夾。

  6. 按一下其中一個資料表資料夾,然後深入查看,直到看到與資料表相關聯的資料。

  7. 按一下代表資料的檔案,然後點選「下載」

  8. 在 Avro 工具 (例如 Avro Viewer) 中開啟這個檔案,確認內容可讀。這表示 Datastream 也將資料轉換為 Avro 檔案格式。

清除所用資源

完成本教學課程後,您可以清除在 Datastream 上建立的資源,這樣資源就不會占用配額,您日後也無須為其付費。下列各節將說明如何刪除或停用這些資源。

刪除專案

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

如要刪除專案:

  1. In the Google Cloud console, go to the Manage resources page.

    Go to Manage resources

  2. In the project list, select the project that you want to delete, and then click Delete.
  3. In the dialog, type the project ID, and then click Shut down to delete the project.

刪除 Cloud Storage 目標值區

  1. 在 Cloud Storage 左側的導覽匣中,按一下「Browser」項目。

  2. 勾選值區左側的核取方塊,然後點選「刪除」

  3. 在「Delete bucket?」(刪除值區?) 視窗中,在文字欄位中輸入值區名稱,然後按一下「CONFIRM」

刪除串流

  1. 確認 Cloud Shell 應用程式已啟用。

  2. 在提示中輸入下列指令:

    curl -X DELETE -H "Authorization: Bearer $TOKEN" -H "Content-Type: application/json" https://datastream.googleapis.com/DATASTREAM_API_VERSION/PROJECT_PATH/streams/STREAM_ID
  3. 確認您看到以下程式碼行:

    {
      "name": "PROJECT_PATH/operations/operation-STREAM_OPERATION_ID",
      "metadata": {
        "@type": "type.googleapis.com/google.cloud.datastream.DATASTREAM_API_VERSION.OperationMetadata",
        "createTime": "DATE_AND_TIME_STAMP",
        "target": "PROJECT_PATH/streams/STREAM_ID",
        "verb": "delete",
        "requestedCancellation": false,
        "apiVersion": "DATASTREAM_API_VERSION"
      },
      "done": false
    }
  4. 確認已刪除串流。在提示中輸入下列指令:

    curl -H "Authorization: Bearer $TOKEN" -H "Content-Type: application/json" https://datastream.googleapis.com/DATASTREAM_API_VERSION/PROJECT_PATH/streams
  5. 確認系統傳回空值 {} 值。這表示 Datastream 中沒有任何串流,且您建立的串流已遭到刪除。

刪除連線設定檔

  1. 刪除來源 Oracle 資料庫的連線設定檔。在提示中輸入下列指令:

    curl -X DELETE -H "Authorization: Bearer $TOKEN" -H "Content-Type: application/json" https://datastream.googleapis.com/DATASTREAM_API_VERSION/PROJECT_PATH/connectionProfiles/SOURCE_CONNECTION_PROFILE_ID
  2. 確認您看到以下程式碼行:

    {
      "name": "PROJECT_PATH/operations/operation-SOURCE_CONNECTION_PROFILE_OPERATION_ID",
      "metadata": {
        "@type": "type.googleapis.com/google.cloud.datastream.DATASTREAM_API_VERSION.OperationMetadata",
        "createTime": "DATE_AND_TIME_STAMP",
        "target": "PROJECT_PATH/connectionProfiles/SOURCE_CONNECTION_PROFILE_ID",
        "verb": "delete",
        "requestedCancellation": false,
        "apiVersion": "DATASTREAM_API_VERSION"
      },
      "done": false
    }
  3. 刪除 Cloud Storage 中目標值區的連線設定檔。在提示中輸入下列指令:

    curl -X DELETE -H "Authorization: Bearer $TOKEN" -H "Content-Type: application/json" https://datastream.googleapis.com/DATASTREAM_API_VERSION/PROJECT_PATH/connectionProfiles/DESTINATION_CONNECTION_PROFILE_ID
  4. 確認您看到以下程式碼行:

    {
      "name": "PROJECT_PATH/operations/operation-DESTINATION_CONNECTION_PROFILE_OPERATION_ID",
      "metadata": {
        "@type": "type.googleapis.com/google.cloud.datastream.DATASTREAM_API_VERSION.OperationMetadata",
        "createTime": "DATE_AND_TIME_STAMP",
        "target": "PROJECT_PATH/connectionProfiles/DESTINATION_CONNECTION_PROFILE_ID",
        "verb": "delete",
        "requestedCancellation": false,
        "apiVersion": "DATASTREAM_API_VERSION"
      },
      "done": false
    }
  5. 確認已刪除兩個連結設定檔。在提示中輸入下列指令:

    curl -H "Authorization: Bearer $TOKEN" -H "Content-Type: application/json" https://datastream.googleapis.com/DATASTREAM_API_VERSION/PROJECT_PATH/connectionProfiles
  6. 確認系統傳回空值 {} 值。這表示 Datastream 中沒有任何連線設定檔,且您建立的設定檔已遭到刪除。

後續步驟

  • 進一步瞭解 Datastream
  • 歡迎自行試用其他 Google Cloud 功能。請參考我們的教學課程