Amazon S3 转移作业

控制台

1. 转到 Google Cloud 控制台中的 BigQuery 页面。

   转到 BigQuery 页面

2. 点击转移作业。

3. 点击创建转移作业。

4. 在创建转移作业页面上：

   • 在来源类型部分的来源中，选择 Amazon S3。

     

   • 在转移配置名称部分的显示名中，输入转移作业的名称，例如 My Transfer。转移作业名称可以是任何容易辨识的值，方便您以后在需要修改该作业时能轻松识别。

     

   • 在时间表选项部分中，执行以下操作：

     • 选择重复频率。如果您选择小时、天、周或月，则还必须指定频率。您还可以选择自定义创建更具体的重复频率。如果您选择按需，则只有当您手动触发转移作业时，此转移作业才会运行。

     • 如果适用，请选择立即开始或在设置的时间开始，并提供开始日期和运行时间。

   • 在 Destination settings 部分的 Destination dataset 中，选择您创建的用来存储数据的数据集。

     

   • 在数据源详细信息部分，执行以下操作：

     • 在 Destination table 部分，输入您创建用来在 BigQuery 中存储数据的表的名称。目标表名称支持使用参数。
     • 在 Amazon S3 URI 部分，按照以下格式输入 URI：s3://mybucket/myfolder/...。URI 也支持使用参数。
     • 在 Access key ID 部分，输入访问密钥 ID。
     • 在私有访问密钥 部分，输入私有访问密钥。
     • 在文件格式部分，选择数据格式：JSON（以换行符分隔）、CSV、Avro、Parquet 或 ORC。

     • 对于写入处置方式，请选择：

       • WRITE_APPEND，以采用增量方式将新数据附加到现有目标表。WRITE_APPEND 是“写入偏好设置”的默认值。
       • WRITE_TRUNCATE，以在每次转移作业运行期间覆盖目标表中的数据。

       如需详细了解 BigQuery Data Transfer Service 如何使用 WRITE_APPEND 或 WRITE_TRUNCATE 注入数据，请参阅 Amazon S3 转移作业的数据注入。如需详细了解 writeDisposition 字段，请参阅 JobConfigurationLoad。

       

   • 在转移作业 - 所有格式部分中：

     • 在允许的错误数部分，输入一个代表可以忽略的错误记录数上限的整数值。
     • （可选）在小数目标类型 (Decimal target types) 部分，输入一个源小数值可转换为的可能 SQL 数据类型的英文逗号分隔列表。选择用于转换的 SQL 数据类型取决于以下条件：
       • 选择用于转换的数据类型将是以下列表中第一个支持源数据的精度和比例的数据类型（按顺序）：NUMERIC、BIGNUMERIC 和 STRING。
       • 如果列出的数据类型均不支持精度和比例，则系统会选择支持指定列表中最宽泛的范围的数据类型。如果在读取源数据时值超出支持的范围，则会抛出错误。
       • 数据类型 STRING 支持所有精度和比例值。
       • 如果将此字段留空，则对于 ORC，数据类型默认为“NUMERIC,STRING”，对于其他文件格式，数据类型默认为“NUMERIC”。
       • 此字段不能包含重复的数据类型。
       • 系统会忽略您在此字段中列出的数据类型的顺序。

     

   • 如果您选择了 CSV 或 JSON 作为文件格式，请在 JSON、CSV 部分中，选中忽略未知值以接受包含与架构不匹配的值所在的行。未知值会被忽略。对于 CSV 文件，此选项会忽略行尾的额外值。

     

   • 如果您选择了 CSV 作为文件格式，请在 CSV 部分中输入用于加载数据的任何其他 CSV 选项。

     

   • 在服务账号菜单中，从与您的 Google Cloud 项目关联的服务账号中选择一个服务账号。您可以将服务账号与转移作业相关联，而不是使用用户凭据。如需详细了解如何将服务账号用于数据转移，请参阅使用服务账号。

     • 如果您使用联合身份登录，则需要有服务账号才能创建转移作业。如果您使用 Google 账号登录，则转移作业的服务账号是可选的。
     • 服务账号必须具有所需的权限。

   • （可选）在通知选项部分，执行以下操作：

     • 点击切换开关以启用电子邮件通知。启用此选项后，转移作业管理员会在转移作业运行失败时收到电子邮件通知。
     • 在选择 Pub/Sub 主题部分，选择您的主题名称，或点击创建主题来创建一个主题。此选项用于为您的转移作业配置 Pub/Sub 运行通知。

5. 点击保存。

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”。

  • 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 来代替缺少的值。

例如，以下命令使用 data_path 值 s3://mybucket/myfile/*.csv、目标数据集 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());
    }
  }
}