將 Amazon S3 資料載入 BigQuery

主控台

1. 前往 Google Cloud 控制台的「資料移轉」頁面。

   前往「資料移轉」

2. 按一下 add「建立移轉作業」。

3. 在「Create Transfer」(建立轉移作業) 頁面：

   • 在「Source type」(來源類型) 部分，「Source」(來源) 請選取「Amazon S3」。

     

   • 在「Transfer config name」(轉移設定名稱) 部分，「Display name」(顯示名稱) 請輸入移轉作業的名稱，例如 My Transfer。移轉作業名稱可以是任意值，日後需要修改移轉作業時能夠據此識別。

     

   • 在「Schedule options」(排程選項) 專區：

     • 選取重複頻率。如果選取「Hours」(小時)、「Days」(天)、「Weeks」(週)或「Months」(月)，必須一併指定頻率。您也可以選取「Custom」(自訂)，建立專屬的重複頻率。如果選取「On-demand」(隨選)，這項資料移轉作業會在您手動觸發後執行。

     • 視情況選取「Start now」(立即開始) 或「Start at set time」(在所設時間開始執行)，並提供開始日期和執行時間。

   • 在「Destination settings」(目的地設定) 部分，「Destination dataset」(目的地資料集) 請選取您為了儲存資料而建立的資料集。

     

   • 在「Data source details」(資料來源詳細資料) 區段：

     • 在「Destination table」(目的地資料表)，輸入您為了在 BigQuery 儲存資料而建立的資料表名稱。目的地資料表的名稱支援參數。
     • 在「Amazon S3 URI」以 s3://mybucket/myfolder/... 的格式輸入 URI (支援參數)。
     • 在「Access key ID」(存取金鑰 ID) 輸入您的存取金鑰 ID。
     • 在「Secret access key」(私密存取金鑰) 輸入您的私密存取金鑰。
     • 在「File format」(檔案格式) 選取資料格式：JSON (以換行符號分隔)、CSV、Avro、Parquet 或 ORC。
     • 在「Write Disposition」(寫入配置) 選取下列其中一個選項：
       • 「WRITE_APPEND」：逐步將新的資料附加至現有目的地資料表。「Write preference」(寫入偏好設定) 的預設值為 WRITE_APPEND。
       • 「WRITE_TRUNCATE」：每次移轉資料時，覆寫目的地資料表資料。

     如要進一步瞭解 BigQuery 資料移轉服務如何使用 WRITE_APPEND 或 WRITE_TRUNCATE 擷取資料，請參閱 Amazon S3 移轉作業資料擷取的相關說明。如要進一步瞭解 writeDisposition 欄位，請參閱 JobConfigurationLoad。

     

   • 在「Transfer Options - All Formats」(移轉選項 - 所有格式) 部分執行下列操作：

     • 在「Number of errors allowed」(允許的錯誤數量) 部分，輸入可以忽略的損壞記錄數量上限 (整數值)。
     • (選用步驟) 在「Decimal target types」(小數目標類型) 部分，輸入以半形逗號分隔的清單，內含來源小數值可能轉換成的 SQL 資料類型。系統會依據下列條件，選取要轉換的 SQL 資料類型：
       • 系統會按照 NUMERIC、BIGNUMERIC 和 STRING 的順序，選取下列清單中第一個支援來源資料有效位數和小數位數的資料類型，做為要轉換的資料類型。
       • 如果清單中的資料類型都不支援有效位數和小數位數，則會選取指定清單中支援範圍最廣的資料類型。如果讀取來源資料時，值超過支援的範圍，就會擲回錯誤。
       • 資料類型 STRING 支援所有有效位數和小數位數值。
       • 如果將這個欄位留空，ORC 的預設資料類型為「NUMERIC, STRING」(NUMERIC、STRING)，其他檔案格式則為「NUMERIC」。
       • 這個欄位不得含有重複的資料類型。
       • 您在這個欄位提供資料類型時採用的順序不會有影響。

     

   • 如果您選取的檔案格式為 CSV 或 JSON，請在「JSON, CSV」(JSON、CSV) 部分勾選「Ignore unknown values」(略過不明的值)，接受所含值不符合結構定義的資料列。系統會忽略不明的值。對於 CSV 檔案，此選項會忽略每行結尾處額外的值。

     

   • 如果您選取的檔案格式為 CSV，請在「CSV」部分針對要載入的資料輸入額外的 CSV 選項。

     

   • 在「Service Account」(服務帳戶) 選單，選取與您的Google Cloud 專案相關聯的服務帳戶。您可以將服務帳戶與資料移轉作業建立關聯，這樣就不需要使用者憑證。如要進一步瞭解如何搭配使用服務帳戶與資料移轉作業，請參閱使用服務帳戶的相關說明。

     • 如果使用聯合身分登入，您必須擁有服務帳戶才能建立資料移轉作業。如果是以 Google 帳戶登入，則不一定要透過服務帳戶建立資料移轉作業。
     • 服務帳戶必須具備必要權限。

   • (選用步驟) 在「Notification options」(通知選項) 部分執行下列操作：

     • 點選切換按鈕，啟用電子郵件通知。啟用這個選項之後，若資料移轉失敗，移轉作業管理員就會收到電子郵件通知。
     • 在「Select a Pub/Sub topic」(選取 Pub/Sub 主題) 選取主題名稱，或是點選「Create a topic」(建立主題)。這個選項會針對資料移轉作業設定 Pub/Sub 執行通知。

4. 按一下 [儲存]。

bq

輸入 bq mk 指令並提供移轉建立標記 - --transfer_config。

bq mk \
--transfer_config \
--project_id=project_id \
--data_source=data_source \
--display_name=name \
--target_dataset=dataset \
--service_account_name=service_account \
--params='parameters'

其中：

• project_id：選用。您的 Google Cloud 專案 ID。 如未提供 --project_id 指定特定專案，系統會使用預設專案。
• data_source：必填。資料來源：amazon_s3。
• display_name：必填。資料移轉設定的顯示名稱。移轉作業名稱可以是任意值，日後需要修改移轉作業時，能夠據此識別即可。
• dataset：必填。資料移轉設定的目標資料集。
• service_account：用於驗證資料移轉的服務帳戶名稱。服務帳戶應由用於建立資料移轉的 project_id 擁有，且應具備所有必要權限。

• parameters：必填。已建立移轉設定的 JSON 格式參數。例如：--params='{"param":"param_value"}'。以下是 Amazon S3 移轉作業的參數：

  • destination_table_name_template：必填。目的地資料表的名稱。

  • data_path：必填。Amazon S3 URI，格式如下：

    s3://mybucket/myfolder/...

    URI 也支援參數。

  • access_key_id：必填。存取金鑰 ID。

  • secret_access_key：必填。您的私密存取金鑰。

  • file_format：選用。指出要轉移的檔案類型：CSV、JSON、AVRO、PARQUET 或 ORC。預設值為 CSV。

  • write_disposition：選用。WRITE_APPEND 只會轉移上次成功執行後修改的檔案。WRITE_TRUNCATE 會轉移所有相符的檔案，包括先前執行轉移作業時轉移的檔案。預設值為 WRITE_APPEND。

  • max_bad_records：選用。允許的損壞記錄數量。預設值為 0。

  • decimal_target_types：選用。以半形逗號分隔的清單，內含來源小數值可能轉換成的 SQL 資料類型。如果未提供這個欄位，ORC 的預設資料類型為「NUMERIC, STRING」(NUMERIC、STRING)，其他檔案格式則為「NUMERIC」。

  • ignore_unknown_values：選填，如果 file_format 不是 JSON 或 CSV，系統會忽略這個值。是否要忽略資料中的不明值。

  • field_delimiter：選用，僅適用於 file_format 為 CSV 的情況。分隔欄位的字元。預設值為逗號。

  • skip_leading_rows：選用，僅適用於 file_format 為 CSV 的情況。指出您不想匯入的標題列數。預設值為 0。

  • allow_quoted_newlines：選用，僅適用於 file_format 為 CSV 的情況。指出是否允許在引用欄位中使用換行符號。

  • allow_jagged_rows：選用，僅適用於 file_format 為 CSV 的情況。指出是否接受缺少結尾自選欄的資料列。缺少的值將以 NULL 填入。

舉例來說，下列指令會使用 s3://mybucket/myfile/*.csv 的 data_path 值、目標資料集 mydataset 以及 file_format CSV，建立名為 My Transfer 的 Amazon S3 資料移轉作業。本範例包含有關 CSV file_format 的選用參數非預設值。

資料移轉作業會在預設專案中建立：

bq mk --transfer_config \
--target_dataset=mydataset \
--display_name='My Transfer' \
--params='{"data_path":"s3://mybucket/myfile/*.csv",
"destination_table_name_template":"MyTable",
"file_format":"CSV",
"write_disposition":"WRITE_APPEND",
"max_bad_records":"1",
"ignore_unknown_values":"true",
"field_delimiter":"|",
"skip_leading_rows":"1",
"allow_quoted_newlines":"true",
"allow_jagged_rows":"false"}' \
--data_source=amazon_s3

執行指令後，您會收到如下的訊息：

[URL omitted] Please copy and paste the above URL into your web browser and follow the instructions to retrieve an authentication code.

請按照指示進行操作，並在指令列中貼上驗證碼。

API

請使用 projects.locations.transferConfigs.create 方法，並提供 TransferConfig 資源的執行個體。

Java

import com.google.api.gax.rpc.ApiException;
import com.google.cloud.bigquery.datatransfer.v1.CreateTransferConfigRequest;
import com.google.cloud.bigquery.datatransfer.v1.DataTransferServiceClient;
import com.google.cloud.bigquery.datatransfer.v1.ProjectName;
import com.google.cloud.bigquery.datatransfer.v1.TransferConfig;
import com.google.protobuf.Struct;
import com.google.protobuf.Value;
import java.io.IOException;
import java.util.HashMap;
import java.util.Map;

// Sample to create amazon s3 transfer config.
public class CreateAmazonS3Transfer {

  public static void main(String[] args) throws IOException {
    // TODO(developer): Replace these variables before running the sample.
    final String projectId = "MY_PROJECT_ID";
    String datasetId = "MY_DATASET_ID";
    String tableId = "MY_TABLE_ID";
    // Amazon S3 Bucket Uri with read role permission
    String sourceUri = "s3://your-bucket-name/*";
    String awsAccessKeyId = "MY_AWS_ACCESS_KEY_ID";
    String awsSecretAccessId = "AWS_SECRET_ACCESS_ID";
    String sourceFormat = "CSV";
    String fieldDelimiter = ",";
    String skipLeadingRows = "1";
    Map<String, Value> params = new HashMap<>();
    params.put(
        "destination_table_name_template", Value.newBuilder().setStringValue(tableId).build());
    params.put("data_path", Value.newBuilder().setStringValue(sourceUri).build());
    params.put("access_key_id", Value.newBuilder().setStringValue(awsAccessKeyId).build());
    params.put("secret_access_key", Value.newBuilder().setStringValue(awsSecretAccessId).build());
    params.put("source_format", Value.newBuilder().setStringValue(sourceFormat).build());
    params.put("field_delimiter", Value.newBuilder().setStringValue(fieldDelimiter).build());
    params.put("skip_leading_rows", Value.newBuilder().setStringValue(skipLeadingRows).build());
    TransferConfig transferConfig =
        TransferConfig.newBuilder()
            .setDestinationDatasetId(datasetId)
            .setDisplayName("Your Aws S3 Config Name")
            .setDataSourceId("amazon_s3")
            .setParams(Struct.newBuilder().putAllFields(params).build())
            .setSchedule("every 24 hours")
            .build();
    createAmazonS3Transfer(projectId, transferConfig);
  }

  public static void createAmazonS3Transfer(String projectId, TransferConfig transferConfig)
      throws IOException {
    try (DataTransferServiceClient client = DataTransferServiceClient.create()) {
      ProjectName parent = ProjectName.of(projectId);
      CreateTransferConfigRequest request =
          CreateTransferConfigRequest.newBuilder()
              .setParent(parent.toString())
              .setTransferConfig(transferConfig)
              .build();
      TransferConfig config = client.createTransferConfig(request);
      System.out.println("Amazon s3 transfer created successfully :" + config.getName());
    } catch (ApiException ex) {
      System.out.print("Amazon s3 transfer was not created." + ex.toString());
    }
  }
}