使用 Dataflow 將儲存空間中的 Parquet 檔案匯入 Lakehouse 執行階段目錄

您可以使用 Dataflow 工作建構工具藍圖，將雲端儲存空間 (Cloud Storage、Amazon S3 或 Azure Blob 儲存空間) 中的現有 Apache Parquet 檔案，新增至無邊界 Lakehouse 中的 Apache Iceberg 資料表。

這個程序會使用 IcebergAddFiles 轉換。如果 Parquet 檔案位於 Cloud Storage 中，這項轉換會向 Lakehouse 註冊檔案，而不會移動或重新編寫基礎資料。如果檔案位於 Amazon S3 等外部儲存系統，系統會將檔案複製到 Cloud Storage，以便透過 Lakehouse 更快速地查詢，然後註冊檔案。

使用下列連線詳細資料，將雲端儲存空間中的 Parquet 檔案新增至 Lakehouse 的 Apache Iceberg 資料表。

事前準備

  1. 啟用 Dataflow、BigQuery 和 Lakehouse API。

  2. 如要取得建立資源所需的權限，請要求管理員授予您專案的必要 Identity and Access Management (IAM) 角色。

  3. 建立 Lakehouse 目錄、命名空間和資料表，以便匯入資料。

  4. 建立雲端儲存空間 bucket (Cloud Storage、Amazon S3 或 Azure Blob Storage)，並將 Parquet 檔案上傳至 bucket。

  5. 如果您使用的雲端儲存空間 bucket 不是 Google Cloud Storage，請建立 Cloud Storage bucket 來儲存工作錯誤記錄。

支援與限制

使用 Dataflow 將雲端儲存空間中的 Parquet 檔案匯入 Lakehouse 時，有下列限制：

  • 來源資料必須為 Apache Parquet 格式，並儲存在 Cloud Storage、Amazon S3 或 Azure Blob 儲存空間。
  • 這項功能僅支援批次管道。

將 Parquet 檔案匯入 Lakehouse

請按照下列步驟操作，使用 Dataflow 工作建構工具 UI，將雲端儲存空間中的 Parquet 檔案匯入 Lakehouse 的 Iceberg 表格。

  1. 前往 Google Cloud 控制台的「Lakehouse」Lakehouse頁面。

    前往 Lakehouse

  2. 選取要匯入資料的目錄、命名空間和資料表。

  3. 在「Table details」(資料表詳細資料) 頁面中，按一下「Import table」(匯入資料表)，然後選取「From Apache Parquet files (Batch)」(從 Apache Parquet 檔案匯入 (批次))。

    系統隨即會開啟 Dataflow 的「Job builder」(工作建立工具) 頁面。

  4. 在「來源」部分：

    1. 開啟已建立的 CreateGlobalInput 來源項目。

    2. 在「YAML 來源設定」編輯器部分，以 elements 序列輸入一或多個 Parquet 檔案的路徑。

      如要提高匯入效率，註冊大量檔案時，請指定多組檔案 (glob)。例如：

      reshuffle: true
elements:
  -   gs://BUCKET_NAME/restaurant-data/2023/*.parquet
  -   gs://BUCKET_NAME/restaurant-data/2024/*.parquet

    3. 按一下「完成」。

  5. 在「Transforms」(轉換) 部分：

    1. 按一下「IcebergAddFiles」IcebergAddFiles轉換區段，開啟該區段。

    2. 在「Iceberg table」(Iceberg 資料表) 欄位中，輸入命名空間和資料表名稱。例如：NAMESPACE .TABLE_NAME .

    3. 在「目錄屬性」下方，設定下列項目：

      1. warehouse：目錄的 Cloud Storage 位置。 例如：gs://CATALOG_PATH。

      2. header.x-goog-user-project：您的 Google Cloud 專案 ID： PROJECT_ID。

      3. 按一下「完成」。

    4. 如果您要從 Amazon S3 或 Azure Blob 儲存體遷移，則需提供額外設定，將 Parquet 檔案複製到 Cloud Storage。如果檔案已在 Cloud Storage 中，則不需要執行這項操作。

      1. 按一下「CopyFilesToGCS」CopyFilesToGCS轉換區段即可開啟。

      2. 設定 gcs_file_path 設定參數的值，提供要複製暫時檔案的完整 Cloud Storage bucket。建議使用 Lakehouse 倉儲所用的 Cloud Storage 值區。

      3. 按一下「完成」。

      4. 按一下「資料流選項」部分，將其展開。

      5. 如果 Parquet 檔案位於 Amazon S3，請按一下「add additional pipeline options」(新增其他管道選項)，提供與 S3 相關的 Apache Beam 管道選項。例如 s3_region_name、s3_access_key_id、s3_secret_access_key 和對應的值。

      6. 如果 Parquet 檔案位於 Azure Blob 儲存空間，請點選「新增其他管道選項」，提供與 Azure 相關的 Apache Beam 管道選項。 例如 azure_connection_string、blob_service_endpoint、azure_managed_identity_client_id，以及對應的值。

  6. 在「Sinks」(接收器) 部分中：

    1. 按一下「Write results」接收器即可開啟。

    2. 在「JSON 位置」欄位中，指定要寫入錯誤結果的 Cloud Storage 位置和檔案名稱。例如：

      gs://BUCKET_NAME/errors/errors.json

    3. 按一下「完成」。

  7. 在「Dataflow Options」部分，按一下「Run job」。

如要進一步自訂用於註冊 Parquet 檔案的 Dataflow 管道，可以使用工作建構工具表單或 YAML 編輯器。

檢查工作輸出內容

工作完成後，您可以在 BigQuery 中查詢 Iceberg 資料表，確認資料已註冊。

  1. 在 Dataflow 工作清單中，確認工作狀態為「成功」。

    前往工作

  2. 如果工作失敗或發生錯誤，請查看 Cloud Storage 中的 JSON 錯誤記錄檔，瞭解詳細資料。

    前往「Buckets」(值區) 頁面

  3. 前往 Google Cloud 控制台的「BigQuery Studio」頁面。

    前往「BigQuery」

  4. 在查詢編輯器中輸入 SQL 查詢，檢查資料表。您可以使用 PROJECT_ID.CATALOG.NAMESPACE.TABLE_NAME 慣例查詢。

    SELECT * FROM `PROJECT_ID.CATALOG.NAMESPACE.TABLE_NAME` LIMIT 10

  5. 按一下「執行」。

  6. 查看「查詢結果」，確認資料已正確處理。

後續步驟