管理转移作业

本文档介绍了如何管理现有数据转移配置。

您还可以手动触发现有转移作业(也称为启动回填运行)。

查看转移作业

通过查看有关每个转移作业的信息、列出所有现有转移作业以及查看转移作业运行历史记录或日志消息,查看现有的转移配置。

所需的角色

如需获得查看转移作业详细信息所需的权限,请让管理员在项目上为您授予 BigQuery User (roles/bigquery.user) IAM 角色。 如需详细了解如何授予角色,请参阅管理对项目、文件夹和组织的访问权限

您也可以通过自定义角色或其他预定义角色来获取所需的权限。

此外,如需通过 Google Cloud 控制台查看日志消息,您必须具有查看 Cloud Logging 数据的权限。Logs Viewer 角色 (roles/logging.viewer) 为您提供对 Logging 所有功能的只读权限。如需详细了解云日志数据所适用的 Identity and Access Management (IAM) 权限和角色,请参阅 Cloud Logging 访问权限控制指南

如需详细了解 BigQuery Data Transfer Service 中的 IAM 角色,请参阅访问权限控制

获取转移作业详细信息

创建传输后,您可以获取有关传输配置的信息。配置包含您在创建转移作业时提供的值,以及资源名称等其他重要信息。

获取转移作业配置相关信息的方法如下:

控制台

  1. 进入数据传输页面。

    转到“数据传输”

  2. 选择要获取详情的转移作业。

  3. 如需查看转移作业配置和数据源详细信息,请点击转移详情页面上的配置。以下示例展示了 Google Ads 转移作业的配置属性:

    控制台中的转移作业配置

bq

输入 bq show 命令并提供转移作业配置的资源名称。--format 标志可用于控制输出格式。

bq show \
--format=prettyjson \
--transfer_config resource_name

resource_name 替换为转移作业的资源名称(也称为转移配置)。如果您不知道转移作业的资源名称,您可以通过以下命令查找资源名称:bq ls --transfer_config --transfer_location=location

例如,输入以下命令可显示转移作业配置 projects/myproject/locations/us/transferConfigs/1234a123-1234-1a23-1be9-12ab3c456de7

bq show \
--format=prettyjson \
--transfer_config projects/myproject/locations/us/transferConfigs/1234a123-1234-1a23-1be9-12ab3c456de7

API

使用 projects.locations.transferConfigs.get 方法并通过 name 参数提供转移作业配置。

Java

试用此示例之前,请按照 BigQuery 快速入门:使用客户端库中的 Java 设置说明进行操作。 如需了解详情,请参阅 BigQuery Java API 参考文档

如需向 BigQuery 进行身份验证,请设置应用默认凭据。 如需了解详情,请参阅为客户端库设置身份验证

import com.google.api.gax.rpc.ApiException;
import com.google.cloud.bigquery.datatransfer.v1.DataTransferServiceClient;
import com.google.cloud.bigquery.datatransfer.v1.GetTransferConfigRequest;
import com.google.cloud.bigquery.datatransfer.v1.TransferConfig;
import java.io.IOException;

// Sample to get config info.
public class GetTransferConfigInfo {

  public static void main(String[] args) throws IOException {
    // TODO(developer): Replace these variables before running the sample.
    String configId = "MY_CONFIG_ID";
    // i.e projects/{project_id}/transferConfigs/{config_id}` or
    // `projects/{project_id}/locations/{location_id}/transferConfigs/{config_id}`
    getTransferConfigInfo(configId);
  }

  public static void getTransferConfigInfo(String configId) throws IOException {
    try (DataTransferServiceClient dataTransferServiceClient = DataTransferServiceClient.create()) {
      GetTransferConfigRequest request =
          GetTransferConfigRequest.newBuilder().setName(configId).build();
      TransferConfig info = dataTransferServiceClient.getTransferConfig(request);
      System.out.print("Config info retrieved successfully." + info.getName() + "\n");
    } catch (ApiException ex) {
      System.out.print("config not found." + ex.toString());
    }
  }
}

列出传输配置

列出项目中的所有现有转移配置:

控制台

  1. 在 Google Cloud 控制台中,转到“数据转移”页面。

    转到“数据传输”

  2. 如果项目中有转移配置,则数据转移列表上会显示转移配置列表。

bq

要按位置列出一个项目的所有转移作业配置,请输入 bq ls 命令并提供 --transfer_location--transfer_config 标志。您还可以提供 --project_id 标志以指定具体项目。如果未指定 --project_id,则系统会使用默认项目。--format 标志可用于控制输出格式。

要列出特定数据源的转移作业配置,请提供 --filter 标志。

要以分页格式查看特定数量的转移作业配置,请提供 --max_results 标志以指定转移作业数量。该命令使用 --page_token 标志返回您提供的页面标记,以查看后续 n 个配置。 如果省略 --max_results,则系统最多会返回 1000 个配置。此外,--max_results 不接受大于 1000 的值。如果您的项目拥有超过 1000 个配置,您可以使用 --max_results--page_token 遍历所有配置。

bq ls \
--transfer_config \
--transfer_location=location \
--project_id=project_id \
--max_results=integer \
--filter=dataSourceIds:data_sources

替换以下内容:

示例:

输入以下命令以显示您的默认项目位于美国的所有转移作业配置。--format 标志用于控制输出。

bq ls \
--format=prettyjson \
--transfer_config \
--transfer_location=us

输入以下命令以显示 ID 为 myproject 的项目位于美国的所有转移作业配置。

bq ls \
--transfer_config \
--transfer_location=us \
--project_id=myproject

输入以下命令以列出最新的 3 个转移作业配置。

bq ls \
--transfer_config \
--transfer_location=us \
--project_id=myproject \
--max_results=3

该命令会返回下一页标记。复制页面标记并在 bq ls 命令中提供此标记,以查看接下来的 3 个结果。

bq ls \
--transfer_config \
--transfer_location=us \
--project_id=myproject \
--max_results=3 \
--page_token=AB1CdEfg_hIJKL

输入以下命令以列出项目 ID 为 myproject 的 Ads 和 Campaign Manager 转移作业配置。

bq ls \
--transfer_config \
--transfer_location=us \
--project_id=myproject \
--filter=dataSourceIds:dcm_dt,google_ads

API

使用 projects.locations.transferConfigs.list 方法并通过 parent 参数提供项目 ID。

Java

试用此示例之前,请按照 BigQuery 快速入门:使用客户端库中的 Java 设置说明进行操作。 如需了解详情,请参阅 BigQuery Java API 参考文档

如需向 BigQuery 进行身份验证,请设置应用默认凭据。 如需了解详情,请参阅为客户端库设置身份验证

import com.google.api.gax.rpc.ApiException;
import com.google.cloud.bigquery.datatransfer.v1.DataTransferServiceClient;
import com.google.cloud.bigquery.datatransfer.v1.ListTransferConfigsRequest;
import com.google.cloud.bigquery.datatransfer.v1.ProjectName;
import java.io.IOException;

// Sample to get list of transfer config
public class ListTransferConfigs {

  public static void main(String[] args) throws IOException {
    // TODO(developer): Replace these variables before running the sample.
    final String projectId = "MY_PROJECT_ID";
    listTransferConfigs(projectId);
  }

  public static void listTransferConfigs(String projectId) throws IOException {
    try (DataTransferServiceClient dataTransferServiceClient = DataTransferServiceClient.create()) {
      ProjectName parent = ProjectName.of(projectId);
      ListTransferConfigsRequest request =
          ListTransferConfigsRequest.newBuilder().setParent(parent.toString()).build();
      dataTransferServiceClient
          .listTransferConfigs(request)
          .iterateAll()
          .forEach(config -> System.out.print("Success! Config ID :" + config.getName() + "\n"));
    } catch (ApiException ex) {
      System.out.println("Config list not found due to error." + ex.toString());
    }
  }
}

Python

试用此示例之前,请按照 BigQuery 快速入门:使用客户端库中的 Python 设置说明进行操作。 如需了解详情,请参阅 BigQuery Python API 参考文档

如需向 BigQuery 进行身份验证,请设置应用默认凭据。 如需了解详情,请参阅为客户端库设置身份验证

from google.cloud import bigquery_datatransfer

transfer_client = bigquery_datatransfer.DataTransferServiceClient()

project_id = "my-project"
parent = transfer_client.common_project_path(project_id)

configs = transfer_client.list_transfer_configs(parent=parent)
print("Got the following configs:")
for config in configs:
    print(f"\tID: {config.name}, Schedule: {config.schedule}")

查看转移作业运行历史记录

在运行已安排好的传输时,系统会为每个传输配置保留运行历史记录,其中包括成功的传输运行和失败的传输运行。超过 90 天的传输运行会从运行历史记录中自动删除。

查看转移作业配置的运行历史记录的方法如下:

控制台

  1. 在 Google Cloud 控制台中,转到“数据转移”页面。

    转到“数据传输”

  2. 点击数据转移作业列表中的转移作业。

  3. 您将转到所选转移作业的运行历史记录页面。

bq

要列出特定转移作业配置的转移作业运行,请输入 bq ls 命令并提供 --transfer_run 标志。您还可以提供 --project_id 标志以指定具体项目。如果 resource_name 不包含项目信息,则系统会使用 --project_id 值。如果未指定 --project_id,则系统会使用默认项目。--format 标志可用于控制输出格式。

要查看特定数量的转移作业运行,请提供 --max_results 标志。该命令使用 --page_token 标志返回您提供的页面标记,以查看后续 n 个配置。

要根据运行状态列出转移作业运行,请提供 --filter 标志。

bq ls \
--transfer_run \
--max_results=integer \
--transfer_location=location \
--project_id=project_id \
--filter=states:state, ... \
resource_name

替换以下内容:

  • integer 是要返回的结果数。
  • location 是转移配置的位置。位置是在您创建转移作业时指定的。
  • project_id 是您的项目 ID。
  • state, ... 是以下内容之一或是一个以英文逗号分隔的列表:
    • SUCCEEDED
    • FAILED
    • PENDING
    • RUNNING
    • CANCELLED
  • resource_name 是转移作业的资源名称(也称为转移配置)。如果您不知道转移作业的资源名称,您可以通过以下命令查找资源名称:bq ls --transfer_config --transfer_location=location

示例:

输入以下命令,以显示转移作业配置 projects/myproject/locations/us/transferConfigs/1234a123-1234-1a23-1be9-12ab3c456de7 的 3 次最新运行。--format 标志用于控制输出。

bq ls \
--format=prettyjson \
--transfer_run \
--max_results=3 \
--transfer_location=us \
projects/myproject/locations/us/transferConfigs/1234a123-1234-1a23-1be9-12ab3c456de7

该命令会返回下一页标记。复制页面标记并在 bq ls 命令中提供此标记,以查看接下来的 3 个结果。

bq ls \
--format=prettyjson \
--transfer_run \
--max_results=3 \
--page_token=AB1CdEfg_hIJKL \
--transfer_location=us \
projects/myproject/locations/us/transferConfigs/1234a123-1234-1a23-1be9-12ab3c456de7

输入以下命令以显示转移作业配置 projects/myproject/locations/us/transferConfigs/1234a123-1234-1a23-1be9-12ab3c456de7 的所有失败的运行。

bq ls \
--format=prettyjson \
--transfer_run \
--filter=states:FAILED \
--transfer_location=us \
projects/myproject/locations/us/transferConfigs/1234a123-1234-1a23-1be9-12ab3c456de7

API

使用 projects.locations.transferConfigs.runs.list 方法并通过 parent 参数指定项目 ID。

Java

试用此示例之前,请按照 BigQuery 快速入门:使用客户端库中的 Java 设置说明进行操作。 如需了解详情,请参阅 BigQuery Java API 参考文档

如需向 BigQuery 进行身份验证,请设置应用默认凭据。 如需了解详情,请参阅为客户端库设置身份验证

import com.google.api.gax.rpc.ApiException;
import com.google.cloud.bigquery.datatransfer.v1.DataTransferServiceClient;
import com.google.cloud.bigquery.datatransfer.v1.ListTransferRunsRequest;
import java.io.IOException;

// Sample to get run history from transfer config.
public class RunHistory {

  public static void main(String[] args) throws IOException {
    // TODO(developer): Replace these variables before running the sample.
    String configId = "MY_CONFIG_ID";
    // i.e projects/{project_id}/transferConfigs/{config_id}` or
    // `projects/{project_id}/locations/{location_id}/transferConfigs/{config_id}`
    runHistory(configId);
  }

  public static void runHistory(String configId) throws IOException {
    try (DataTransferServiceClient dataTransferServiceClient = DataTransferServiceClient.create()) {
      ListTransferRunsRequest request =
          ListTransferRunsRequest.newBuilder().setParent(configId).build();
      dataTransferServiceClient
          .listTransferRuns(request)
          .iterateAll()
          .forEach(run -> System.out.print("Success! Run ID :" + run.getName() + "\n"));
    } catch (ApiException ex) {
      System.out.println("Run history not found due to error." + ex.toString());
    }
  }
}

查看转移作业运行详情和日志消息

在传输运行出现在运行历史记录中时,您可以查看运行详情,包括日志消息、警告和错误、运行名称以及开始和结束时间。

查看转移作业运行详情的方法如下:

控制台

  1. 在 Google Cloud 控制台中,转到“数据转移”页面。

    转到“数据传输”

  2. 点击数据转移作业列表中的转移作业。

  3. 您将转到所选转移作业的运行历史记录页面。

  4. 点击各次转移作业运行,相应的运行详情面板就会打开。

  5. 运行详情中,记下所有错误消息。在联系 Cloud Customer Care 时,您需要用到这些信息。运行详情还包括日志消息和警告。

    控制台中的“运行详情”

bq

要查看转移作业运行详情,请输入 bq show 命令并使用 --transfer_run 标志提供转移作业运行的名称。--format 标志可用于控制输出格式。

bq show \
--format=prettyjson \
--transfer_run run_name

run_name 替换为转移作业运行的运行名称。您可以使用 bq ls 命令检索运行名称。

示例:

输入以下命令以显示转移作业运行 projects/myproject/locations/us/transferConfigs/1234a123-1234-1a23-1be9-12ab3c456de7/runs/1a2b345c-0000-1234-5a67-89de1f12345g 的详情。

bq show \
--format=prettyjson \
--transfer_run \
projects/myproject/locations/us/transferConfigs/1234a123-1234-1a23-1be9-12ab3c456de7/runs/1a2b345c-0000-1234-5a67-89de1f12345g

要查看转移作业运行的转移作业日志消息,请输入带有 --transfer_log 标志的 bq ls 命令。您可以使用 --message_type 标志按类型过滤日志消息。

要查看特定数量的日志消息,请提供 --max_results 标志。该命令使用 --page_token 标志返回您提供的页面标记,以查看后续 n 条消息。

bq ls \
--transfer_log \
--max_results=integer \
--message_type=messageTypes:message_type \
run_name

替换以下内容:

  • integer 是要返回的日志消息数量。
  • message_type 是要查看的日志消息的类型(单个值或以英文逗号分隔的列表):
    • INFO
    • WARNING
    • ERROR
  • run_name 是转移作业运行名称。您可以使用 bq ls 命令检索运行名称。

示例:

输入以下命令以查看转移作业运行 projects/myproject/locations/us/transferConfigs/1234a123-1234-1a23-1be9-12ab3c456de7/runs/1a2b345c-0000-1234-5a67-89de1f12345g 的前 2 条日志消息。

bq ls \
--transfer_log \
--max_results=2 \
projects/myproject/locations/us/transferConfigs/1234a123-1234-1a23-1be9-12ab3c456de7/runs/1a2b345c-0000-1234-5a67-89de1f12345g

该命令会返回下一页标记。复制页面标记并在 bq ls 命令中提供此标记,以查看接下来的 2 个结果。

bq ls \
--transfer_log \
--max_results=2 \
--page_token=AB1CdEfg_hIJKL \
projects/myproject/locations/us/transferConfigs/1234a123-1234-1a23-1be9-12ab3c456de7/runs/1a2b345c-0000-1234-5a67-89de1f12345g

输入以下命令以仅查看转移作业运行 projects/myproject/locations/us/transferConfigs/1234a123-1234-1a23-1be9-12ab3c456de7/runs/1a2b345c-0000-1234-5a67-89de1f12345g 的错误消息。

bq ls \
--transfer_log \
--message_type=messageTypes:ERROR \
projects/myproject/locations/us/transferConfigs/1234a123-1234-1a23-1be9-12ab3c456de7/runs/1a2b345c-0000-1234-5a67-89de1f12345g

API

使用 projects.transferConfigs.runs.transferLogs.list 方法并通过 parent 参数提供转移作业运行的名称。

Java

试用此示例之前,请按照 BigQuery 快速入门:使用客户端库中的 Java 设置说明进行操作。 如需了解详情,请参阅 BigQuery Java API 参考文档

如需向 BigQuery 进行身份验证,请设置应用默认凭据。 如需了解详情,请参阅为客户端库设置身份验证

import com.google.api.gax.rpc.ApiException;
import com.google.cloud.bigquery.datatransfer.v1.DataTransferServiceClient;
import com.google.cloud.bigquery.datatransfer.v1.GetTransferRunRequest;
import com.google.cloud.bigquery.datatransfer.v1.TransferRun;
import java.io.IOException;

// Sample to get run details from transfer config.
public class RunDetails {

  public static void main(String[] args) throws IOException {
    // TODO(developer): Replace these variables before running the sample.
    // runId examples:
    // `projects/{project_id}/transferConfigs/{config_id}/runs/{run_id}` or
    // `projects/{project_id}/locations/{location_id}/transferConfigs/{config_id}/runs/{run_id}`
    String runId = "MY_RUN_ID";
    runDetails(runId);
  }

  public static void runDetails(String runId) throws IOException {
    try (DataTransferServiceClient dataTransferServiceClient = DataTransferServiceClient.create()) {
      GetTransferRunRequest request = GetTransferRunRequest.newBuilder().setName(runId).build();
      TransferRun run = dataTransferServiceClient.getTransferRun(request);
      System.out.print("Run details retrieved successfully :" + run.getName() + "\n");
    } catch (ApiException ex) {
      System.out.print("Run details not found." + ex.toString());
    }
  }
}

修改转移作业

您可以修改现有转移作业,方法是修改转移配置的相关信息,更新附加到转移配置的用户凭据,以及停用或删除转移作业。

所需的角色

如需获得修改转移作业所需的权限,请让管理员在项目上为您授予 BigQuery Admin (roles/bigquery.admin) IAM 角色。如需详细了解如何授予角色,请参阅管理对项目、文件夹和组织的访问权限

您也可以通过自定义角色或其他预定义角色来获取所需的权限。

更新传输

创建转移配置后,您可以修改以下字段:

  • 目标数据集
  • 显示名称
  • 为特定传输类型指定的任何参数
  • 运行通知设置
  • 服务账号

创建转移作业后,您无法修改转移作业的来源。

更新转移作业的方法如下:

控制台

  1. 在 Google Cloud 控制台中,转到“数据转移”页面。

    转到“数据传输”

  2. 点击数据转移作业列表中的转移作业。

  3. 点击修改以更新转移作业配置。

bq

输入 bq update 命令,使用 --transfer_config 标志提供转移配置的资源名称,并指定 --display_name--params--refresh_window_days--schedule--target_dataset 标志。您可以选择为计划查询Cloud Storage 转移作业指定 --destination_kms_key 标志。

bq update \
--display_name='NAME' \
--params='PARAMETERS' \
--refresh_window_days=INTEGER \
--schedule='SCHEDULE'
--target_dataset=DATASET_ID \
--destination_kms_key="DESTINATION_KEY" \
--transfer_config \
--service_account_name=SERVICE_ACCOUNT \
RESOURCE_NAME

替换以下内容:

  • NAME:转移配置的显示名称。
  • PARAMETERS:转移作业配置的参数(采用 JSON 格式)。例如:--params='{"param1":"param_value1"}'。您可以修改以下参数:
    • Campaign Manager:bucketnetwork_id
    • Google Ad Manager:bucketnetwork_code
    • Google Ads:customer_id
    • Google Merchant Center:merchant_id
    • Google Play:buckettable_suffix
    • 计划查询:destination_table_kms_keydestination_table_name_templatepartitioning_fieldpartitioning_typequerywrite_disposition
    • Search Ads 360:advertiser_idagency_idcustom_floodlight_variablesinclude_removed_entitiestable_filter
    • YouTube 频道:table_suffix
    • YouTube 内容所有者:content_owner_idtable_suffix
  • INTEGER:一个介于 0 到 30 之间的值。如需了解如何设置刷新时段,请参阅对应于您的转移作业类型的文档。
  • SCHEDULE:周期性安排,例如 --schedule="every 3 hours"。如需了解 schedule 语法,请参阅设置 schedule 的格式
  • DATASET_ID:转移作业配置的目标数据集。
  • DESTINATION_KEYCloud KMS 密钥资源 ID,例如 projects/project_name/locations/us/keyRings/key_ring_name/cryptoKeys/key_name。CMEK 仅适用于预定查询Cloud Storage 转移作业。
  • SERVICE_ACCOUNT:指定要用于转移作业的服务账号。
  • RESOURCE_NAME:转移作业的资源名称(也称为转移作业配置)。如果您不知道转移作业的资源名称,您可以通过以下命令查找资源名称:bq ls --transfer_config --transfer_location=location

示例:

以下命令将更新 Google Ads 转移作业 projects/myproject/locations/us/transferConfigs/1234a123-1234-1a23-1be9-12ab3c456de7 的显示名、目标数据集、刷新时段和参数:

bq update \
--display_name='My changed transfer' \
--params='{"customer_id":"123-123-5678"}' \
--refresh_window_days=3 \
--target_dataset=mydataset2 \
--transfer_config \
 projects/myproject/locations/us/transferConfigs/1234a123-1234-1a23-1be9-12ab3c456de7

以下命令将更新计划查询转移作业 projects/myproject/locations/us/transferConfigs/5678z567-5678-5z67-5yx9-56zy3c866vw9 的参数和时间表:

bq update \
--params='{"destination_table_name_template":"test", "write_disposition":"APPEND"}' \
--schedule="every 24 hours" \
--transfer_config \
projects/myproject/locations/us/transferConfigs/5678z567-5678-5z67-5yx9-56zy3c866vw9

API

使用 projects.transferConfigs.patch 方法并通过 transferConfig.name 参数提供转移作业的资源名称。如果您不知道转移作业的资源名称,您可以通过以下命令查找资源名称:bq ls --transfer_config --transfer_location=location。您还可以调用以下方法,并使用 parent 参数提供项目 ID 以列出所有转移作业:projects.locations.transferConfigs.list

Java

试用此示例之前,请按照 BigQuery 快速入门:使用客户端库中的 Java 设置说明进行操作。 如需了解详情,请参阅 BigQuery Java API 参考文档

如需向 BigQuery 进行身份验证,请设置应用默认凭据。 如需了解详情,请参阅为客户端库设置身份验证

import com.google.api.gax.rpc.ApiException;
import com.google.cloud.bigquery.datatransfer.v1.DataTransferServiceClient;
import com.google.cloud.bigquery.datatransfer.v1.TransferConfig;
import com.google.cloud.bigquery.datatransfer.v1.UpdateTransferConfigRequest;
import com.google.protobuf.FieldMask;
import com.google.protobuf.util.FieldMaskUtil;
import java.io.IOException;

// Sample to update transfer config.
public class UpdateTransferConfig {

  public static void main(String[] args) throws IOException {
    // TODO(developer): Replace these variables before running the sample.
    String configId = "MY_CONFIG_ID";
    TransferConfig transferConfig =
        TransferConfig.newBuilder()
            .setName(configId)
            .setDisplayName("UPDATED_DISPLAY_NAME")
            .build();
    FieldMask updateMask = FieldMaskUtil.fromString("display_name");
    updateTransferConfig(transferConfig, updateMask);
  }

  public static void updateTransferConfig(TransferConfig transferConfig, FieldMask updateMask)
      throws IOException {
    try (DataTransferServiceClient dataTransferServiceClient = DataTransferServiceClient.create()) {
      UpdateTransferConfigRequest request =
          UpdateTransferConfigRequest.newBuilder()
              .setTransferConfig(transferConfig)
              .setUpdateMask(updateMask)
              .build();
      TransferConfig updateConfig = dataTransferServiceClient.updateTransferConfig(request);
      System.out.println("Transfer config updated successfully :" + updateConfig.getDisplayName());
    } catch (ApiException ex) {
      System.out.print("Transfer config was not updated." + ex.toString());
    }
  }
}

Python

试用此示例之前,请按照 BigQuery 快速入门:使用客户端库中的 Python 设置说明进行操作。 如需了解详情,请参阅 BigQuery Python API 参考文档

如需向 BigQuery 进行身份验证,请设置应用默认凭据。 如需了解详情,请参阅为客户端库设置身份验证

from google.cloud import bigquery_datatransfer
from google.protobuf import field_mask_pb2

transfer_client = bigquery_datatransfer.DataTransferServiceClient()

transfer_config_name = "projects/1234/locations/us/transferConfigs/abcd"
new_display_name = "My Transfer Config"

transfer_config = bigquery_datatransfer.TransferConfig(name=transfer_config_name)
transfer_config.display_name = new_display_name

transfer_config = transfer_client.update_transfer_config(
    {
        "transfer_config": transfer_config,
        "update_mask": field_mask_pb2.FieldMask(paths=["display_name"]),
    }
)

print(f"Updated config: '{transfer_config.name}'")
print(f"New display name: '{transfer_config.display_name}'")

更新凭据

传输使用创建该传输的用户的凭据。如果您需要更改附加到某项传输配置的用户,则可以更新传输的凭据。如果创建传输的用户不再与贵组织合作,这种做法非常有用。

更新转移作业凭据的方法如下:

控制台

  1. 在 Google Cloud 控制台中,以您要转让所有权的用户身份登录。

  2. 前往“数据转移作业”页面。

    转到“数据传输”

  3. 点击数据转移作业列表中的转移作业。

  4. 点击更多菜单,然后选择刷新凭据 (Refresh credentials)。

  5. 点击允许,授予 BigQuery Data Transfer Service 权限,使其能够查看您的报告数据以及在 BigQuery 中访问和管理数据。

bq

输入 bq update 命令,使用 --transfer_config 标志提供转移作业配置的资源名称,并指定 --update_credentials 标志。

bq update \
--update_credentials=boolean \
--transfer_config \
resource_name

替换以下内容:

  • boolean 是一个布尔值,指示是否应为转移作业配置更新凭据。
  • resource_name 是转移作业的资源名称(也称为转移配置)。如果您不知道转移作业的资源名称,您可以通过以下命令查找资源名称:bq ls --transfer_config --transfer_location=location

示例:

输入以下命令,以更新 Google Ads 转移作业 projects/myproject/locations/us/transferConfigs/1234a123-1234-1a23-1be9-12ab3c456de7 的凭据。

bq update \
--update_credentials=true \
--transfer_config \
projects/myproject/locations/us/transferConfigs/1234a123-1234-1a23-1be9-12ab3c456de7

API

使用 projects.transferConfigs.patch 方法并提供 authorizationCodeupdateMask 参数。

Java

试用此示例之前,请按照 BigQuery 快速入门:使用客户端库中的 Java 设置说明进行操作。 如需了解详情,请参阅 BigQuery Java API 参考文档

如需向 BigQuery 进行身份验证,请设置应用默认凭据。 如需了解详情,请参阅为客户端库设置身份验证

import com.google.api.gax.rpc.ApiException;
import com.google.cloud.bigquery.datatransfer.v1.DataTransferServiceClient;
import com.google.cloud.bigquery.datatransfer.v1.TransferConfig;
import com.google.cloud.bigquery.datatransfer.v1.UpdateTransferConfigRequest;
import com.google.protobuf.FieldMask;
import com.google.protobuf.util.FieldMaskUtil;
import java.io.IOException;

// Sample to update credentials in transfer config.
public class UpdateCredentials {

  public static void main(String[] args) throws IOException {
    // TODO(developer): Replace these variables before running the sample.
    String configId = "MY_CONFIG_ID";
    String serviceAccount = "MY_SERVICE_ACCOUNT";
    TransferConfig transferConfig = TransferConfig.newBuilder().setName(configId).build();
    FieldMask updateMask = FieldMaskUtil.fromString("service_account_name");
    updateCredentials(transferConfig, serviceAccount, updateMask);
  }

  public static void updateCredentials(
      TransferConfig transferConfig, String serviceAccount, FieldMask updateMask)
      throws IOException {
    try (DataTransferServiceClient dataTransferServiceClient = DataTransferServiceClient.create()) {
      UpdateTransferConfigRequest request =
          UpdateTransferConfigRequest.newBuilder()
              .setTransferConfig(transferConfig)
              .setUpdateMask(updateMask)
              .setServiceAccountName(serviceAccount)
              .build();
      dataTransferServiceClient.updateTransferConfig(request);
      System.out.println("Credentials updated successfully");
    } catch (ApiException ex) {
      System.out.print("Credentials was not updated." + ex.toString());
    }
  }
}

Python

试用此示例之前,请按照 BigQuery 快速入门:使用客户端库中的 Python 设置说明进行操作。 如需了解详情,请参阅 BigQuery Python API 参考文档

如需向 BigQuery 进行身份验证,请设置应用默认凭据。 如需了解详情,请参阅为客户端库设置身份验证

from google.cloud import bigquery_datatransfer
from google.protobuf import field_mask_pb2

transfer_client = bigquery_datatransfer.DataTransferServiceClient()

service_account_name = "abcdef-test-sa@abcdef-test.iam.gserviceaccount.com"
transfer_config_name = "projects/1234/locations/us/transferConfigs/abcd"

transfer_config = bigquery_datatransfer.TransferConfig(name=transfer_config_name)

transfer_config = transfer_client.update_transfer_config(
    {
        "transfer_config": transfer_config,
        "update_mask": field_mask_pb2.FieldMask(paths=["service_account_name"]),
        "service_account_name": service_account_name,
    }
)

print("Updated config: '{}'".format(transfer_config.name))

停用传输

停用转移作业时,disabled 会添加到转移作业名称。停用转移作业时,系统不会安排新的转移作业运行,也不允许新的回填。任何正在进行的转移作业运行均会完成。

停用转移作业不会移除已转移到 BigQuery 的任何数据。除非删除数据集删除表,否则您需要为之前转移的数据支付标准 BigQuery 存储费用

停用转移作业的方法如下:

控制台

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

    转到 BigQuery

  2. 点击转移作业

  3. 转移作业页面上,点击列表中要停用的转移作业。

  4. 点击停用。要重新启用转移作业,请点击启用

bq

CLI 不支持停用转移作业。

API

使用 projects.locations.transferConfigs.patch 方法并在 projects.locations.transferConfig 资源中将 disabled 设置为 true

Java

试用此示例之前,请按照 BigQuery 快速入门:使用客户端库中的 Java 设置说明进行操作。 如需了解详情,请参阅 BigQuery Java API 参考文档

如需向 BigQuery 进行身份验证,请设置应用默认凭据。 如需了解详情,请参阅为客户端库设置身份验证

import com.google.api.gax.rpc.ApiException;
import com.google.cloud.bigquery.datatransfer.v1.DataTransferServiceClient;
import com.google.cloud.bigquery.datatransfer.v1.TransferConfig;
import com.google.cloud.bigquery.datatransfer.v1.UpdateTransferConfigRequest;
import com.google.protobuf.FieldMask;
import com.google.protobuf.util.FieldMaskUtil;
import java.io.IOException;

// Sample to disable transfer config.
public class DisableTransferConfig {

  public static void main(String[] args) throws IOException {
    // TODO(developer): Replace these variables before running the sample.
    String configId = "MY_CONFIG_ID";
    TransferConfig transferConfig =
        TransferConfig.newBuilder().setName(configId).setDisabled(true).build();
    FieldMask updateMask = FieldMaskUtil.fromString("disabled");
    disableTransferConfig(transferConfig, updateMask);
  }

  public static void disableTransferConfig(TransferConfig transferConfig, FieldMask updateMask)
      throws IOException {
    try (DataTransferServiceClient dataTransferServiceClient = DataTransferServiceClient.create()) {
      UpdateTransferConfigRequest request =
          UpdateTransferConfigRequest.newBuilder()
              .setTransferConfig(transferConfig)
              .setUpdateMask(updateMask)
              .build();
      TransferConfig updateConfig = dataTransferServiceClient.updateTransferConfig(request);
      System.out.println("Transfer config disabled successfully :" + updateConfig.getDisplayName());
    } catch (ApiException ex) {
      System.out.print("Transfer config was not disabled." + ex.toString());
    }
  }
}

重新启用转移作业:

试用此示例之前,请按照 BigQuery 快速入门:使用客户端库中的 Java 设置说明进行操作。 如需了解详情,请参阅 BigQuery Java API 参考文档

如需向 BigQuery 进行身份验证,请设置应用默认凭据。 如需了解详情,请参阅为客户端库设置身份验证

/*
 * Copyright 2020 Google LLC
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 * http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

package com.example.bigquerydatatransfer;

import com.google.api.gax.rpc.ApiException;
import com.google.cloud.bigquery.datatransfer.v1.DataTransferServiceClient;
import com.google.cloud.bigquery.datatransfer.v1.TransferConfig;
import com.google.cloud.bigquery.datatransfer.v1.UpdateTransferConfigRequest;
import com.google.protobuf.FieldMask;
import com.google.protobuf.util.FieldMaskUtil;
import java.io.IOException;

// Sample to disable transfer config.
public class DisableTransferConfig {

  public static void main(String[] args) throws IOException {
    // TODO(developer): Replace these variables before running the sample.
    String configId = "MY_CONFIG_ID";
    TransferConfig transferConfig =
        TransferConfig.newBuilder().setName(configId).setDisabled(true).build();
    FieldMask updateMask = FieldMaskUtil.fromString("disabled");
    disableTransferConfig(transferConfig, updateMask);
  }

  public static void disableTransferConfig(TransferConfig transferConfig, FieldMask updateMask)
      throws IOException {
    try (DataTransferServiceClient dataTransferServiceClient = DataTransferServiceClient.create()) {
      UpdateTransferConfigRequest request =
          UpdateTransferConfigRequest.newBuilder()
              .setTransferConfig(transferConfig)
              .setUpdateMask(updateMask)
              .build();
      TransferConfig updateConfig = dataTransferServiceClient.updateTransferConfig(request);
      System.out.println("Transfer config disabled successfully :" + updateConfig.getDisplayName());
    } catch (ApiException ex) {
      System.out.print("Transfer config was not disabled." + ex.toString());
    }
  }
}

删除传输

删除传输时,系统不会安排新的传输运行。任何正在进行的传输运行均会停止。

删除传输不会移除已传输至 BigQuery 的任何数据。除非删除数据集删除表,否则您需要为之前转移的数据支付标准 BigQuery 存储费用

删除转移作业的方法如下:

控制台

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

    转到 BigQuery

  2. 点击转移作业

  3. 转移作业页面上,点击列表中要删除的转移作业。

  4. 点击删除。为安全起见,您需要在框中输入“delete”一词以确认您的意图。

bq

输入 bq rm 命令并提供转移作业配置的资源名称。您可以使用 -f 标志在不经确认的情况下删除转移作业配置。

bq rm \
-f \
--transfer_config \
resource_name

其中:

例如,输入以下命令可删除转移作业配置 projects/myproject/locations/us/transferConfigs/1234a123-1234-1a23-1be9-12ab3c456de7

bq rm \
--transfer_config \
projects/myproject/locations/us/transferConfigs/1234a123-1234-1a23-1be9-12ab3c456de7

API

使用 projects.locations.transferConfigs.delete 方法并通过 name 参数提供要删除的资源。

Java

试用此示例之前,请按照 BigQuery 快速入门:使用客户端库中的 Java 设置说明进行操作。 如需了解详情,请参阅 BigQuery Java API 参考文档

如需向 BigQuery 进行身份验证,请设置应用默认凭据。 如需了解详情,请参阅为客户端库设置身份验证

import com.google.api.gax.rpc.ApiException;
import com.google.cloud.bigquery.datatransfer.v1.DataTransferServiceClient;
import com.google.cloud.bigquery.datatransfer.v1.DeleteTransferConfigRequest;
import java.io.IOException;

// Sample to delete a transfer config
public class DeleteTransferConfig {

  public static void main(String[] args) throws IOException {
    // TODO(developer): Replace these variables before running the sample.
    // i.e projects/{project_id}/transferConfigs/{config_id}` or
    // `projects/{project_id}/locations/{location_id}/transferConfigs/{config_id}`
    String configId = "MY_CONFIG_ID";
    deleteTransferConfig(configId);
  }

  public static void deleteTransferConfig(String configId) throws IOException {
    try (DataTransferServiceClient dataTransferServiceClient = DataTransferServiceClient.create()) {
      DeleteTransferConfigRequest request =
          DeleteTransferConfigRequest.newBuilder().setName(configId).build();
      dataTransferServiceClient.deleteTransferConfig(request);
      System.out.println("Transfer config deleted successfully");
    } catch (ApiException ex) {
      System.out.println("Transfer config was not deleted." + ex.toString());
    }
  }
}

Python

试用此示例之前,请按照 BigQuery 快速入门:使用客户端库中的 Python 设置说明进行操作。 如需了解详情,请参阅 BigQuery Python API 参考文档

如需向 BigQuery 进行身份验证,请设置应用默认凭据。 如需了解详情,请参阅为客户端库设置身份验证

import google.api_core.exceptions
from google.cloud import bigquery_datatransfer

transfer_client = bigquery_datatransfer.DataTransferServiceClient()

transfer_config_name = "projects/1234/locations/us/transferConfigs/abcd"
try:
    transfer_client.delete_transfer_config(name=transfer_config_name)
except google.api_core.exceptions.NotFound:
    print("Transfer config not found.")
else:
    print(f"Deleted transfer config: {transfer_config_name}")

手动触发转移作业

您可以手动触发转移作业(也称为回填运行),以在自动计划的转移作业之外加载其他数据文件。对于支持运行时参数的数据源,您还可以通过指定日期或时间范围来手动触发转移,以从中加载过去的数据。

您可以随时手动启动数据回填。除了源限制之外,BigQuery Data Transfer Service 为每项回填请求提供最长 180 天的支持。该服务不支持多项同时回填请求。

如需了解可以回填的数据量,请参阅您的数据源对应的转移指南。

所需的角色

如需获得修改转移作业所需的权限,请让管理员在项目上为您授予 BigQuery Admin (roles/bigquery.admin) IAM 角色。如需详细了解如何授予角色,请参阅管理对项目、文件夹和组织的访问权限

您也可以通过自定义角色或其他预定义角色来获取所需的权限。

手动触发转移或回填

您可以通过以下方法手动触发转移或回填运行:

  • 使用 Google Cloud 控制台选择转移作业运行,然后点击立即运行转移作业安排回填
  • 通过 bq 命令行工具使用 bq mk –transfer run 命令
  • 调用 projects.locations.transferConfigs.startManualRuns method API 方法

如需了解每种方法的详细说明,请选择相应的标签页:

控制台

  1. 在 Google Cloud 控制台中,转到“数据转移”页面。

    转到“数据传输”

  2. 从列表中选择您的转移作业。

  3. 点击立即运行转移作业安排回填。根据转移配置类型,只有一个选项可用。

    • 如果您点击了立即运行转移作业,请选择运行一次性转移作业在特定日期运行(如适用)。如果您选择了在特定日期运行,请选择特定日期和时间:

      立即运行转移作业

    • 如果您点击了安排回填,请选择运行一次性转移作业在某个日期范围运行(如适用)。如果您选择了在某个日期范围运行,请选择开始和结束日期及时间:

      安排回填

  4. 点击确定

bq

如需手动启动转移作业运行,请输入带 --transfer_run 标志的 bq mk 命令:

bq mk \
--transfer_run \
--run_time='RUN_TIME' \
RESOURCE_NAME

替换以下内容:

  • RUN_TIME 是时间戳,用于指定过去的转移日期。使用以 Z 结尾或包含有效时区偏移值的时间戳,例如 2022-08-19T12:11:35.00Z2022-05-25T00:00:00+00:00
    • 如果您的转移没有运行时参数,或者您只想立即触发转移而不指定过去的转移,请在此字段中输入当前时间。
  • RESOURCE_NAME 是转移配置中列出的资源名称,例如 projects/myproject/locations/us/transferConfigs/1234a123-1234-1a23-1be9-12ab3c456de7

如需手动针对某个日期范围启动转移作业运行,请输入带 --transfer_run 标志的 bq mk 命令以及日期范围:

bq mk \
--transfer_run \
--start_time='START_TIME' \
--end_time='END_TIME' \
RESOURCE_NAME

替换以下内容:

  • START_TIMEEND_TIME 是以 Z 结尾或包含有效时区偏移值的时间戳。这些值指定了包含您要回填的之前转移作业运行的时间范围,例如 2022-08-19T12:11:35.00Z2022-05-25T00:00:00+00:00
  • RESOURCE_NAME 是转移配置中列出的资源名称,例如 projects/myproject/locations/us/transferConfigs/1234a123-1234-1a23-1be9-12ab3c456de7

API

如需手动启动转移作业运行,请使用 projects.locations.transferConfigs.startManualRuns 方法并通过父参数提供转移配置资源名称。要查找转移配置的资源名称,请参阅获取转移作业相关信息

  "requestedRunTime": "RUN_TIME"

替换以下内容:

  • RUN_TIME 是时间戳,用于指定过去的转移日期。使用以 Z 结尾或包含有效时区偏移值的时间戳,例如 2022-08-19T12:11:35.00Z2022-05-25T00:00:00+00:00
    • 如果您的转移没有运行时参数,或者您只想立即触发转移而不指定过去的转移,请在此字段中输入当前时间。

要针对某个日期范围手动启动转移作业运行,请提供日期范围:

"requestedTimeRange": {
  "startTime": "START_TIME",
  "endTime": "END_TIME"
}

替换以下内容:

  • START_TIMEEND_TIME 是以 Z 结尾或包含有效时区偏移值的时间戳。这些值指定了包含您要回填的之前转移作业运行的时间范围,例如 2022-08-19T12:11:35.00Z2022-05-25T00:00:00+00:00

Java

试用此示例之前,请按照 BigQuery 快速入门:使用客户端库中的 Java 设置说明进行操作。 如需了解详情,请参阅 BigQuery Java API 参考文档

如需向 BigQuery 进行身份验证,请设置应用默认凭据。 如需了解详情,请参阅为客户端库设置身份验证

import com.google.api.gax.rpc.ApiException;
import com.google.cloud.bigquery.datatransfer.v1.DataTransferServiceClient;
import com.google.cloud.bigquery.datatransfer.v1.ScheduleTransferRunsRequest;
import com.google.cloud.bigquery.datatransfer.v1.ScheduleTransferRunsResponse;
import com.google.protobuf.Timestamp;
import java.io.IOException;
import org.threeten.bp.Clock;
import org.threeten.bp.Instant;
import org.threeten.bp.temporal.ChronoUnit;

// Sample to run schedule back fill for transfer config
public class ScheduleBackFill {

  public static void main(String[] args) throws IOException {
    // TODO(developer): Replace these variables before running the sample.
    String configId = "MY_CONFIG_ID";
    Clock clock = Clock.systemDefaultZone();
    Instant instant = clock.instant();
    Timestamp startTime =
        Timestamp.newBuilder()
            .setSeconds(instant.minus(5, ChronoUnit.DAYS).getEpochSecond())
            .setNanos(instant.minus(5, ChronoUnit.DAYS).getNano())
            .build();
    Timestamp endTime =
        Timestamp.newBuilder()
            .setSeconds(instant.minus(2, ChronoUnit.DAYS).getEpochSecond())
            .setNanos(instant.minus(2, ChronoUnit.DAYS).getNano())
            .build();
    scheduleBackFill(configId, startTime, endTime);
  }

  public static void scheduleBackFill(String configId, Timestamp startTime, Timestamp endTime)
      throws IOException {
    try (DataTransferServiceClient client = DataTransferServiceClient.create()) {
      ScheduleTransferRunsRequest request =
          ScheduleTransferRunsRequest.newBuilder()
              .setParent(configId)
              .setStartTime(startTime)
              .setEndTime(endTime)
              .build();
      ScheduleTransferRunsResponse response = client.scheduleTransferRuns(request);
      System.out.println("Schedule backfill run successfully :" + response.getRunsCount());
    } catch (ApiException ex) {
      System.out.print("Schedule backfill was not run." + ex.toString());
    }
  }
}

Python

试用此示例之前,请按照 BigQuery 快速入门:使用客户端库中的 Python 设置说明进行操作。 如需了解详情,请参阅 BigQuery Python API 参考文档

如需向 BigQuery 进行身份验证,请设置应用默认凭据。 如需了解详情,请参阅为客户端库设置身份验证

import datetime

from google.cloud.bigquery_datatransfer_v1 import (
    DataTransferServiceClient,
    StartManualTransferRunsRequest,
)

# Create a client object
client = DataTransferServiceClient()

# Replace with your transfer configuration name
transfer_config_name = "projects/1234/locations/us/transferConfigs/abcd"
now = datetime.datetime.now(datetime.timezone.utc)
start_time = now - datetime.timedelta(days=5)
end_time = now - datetime.timedelta(days=2)

# Some data sources, such as scheduled_query only support daily run.
# Truncate start_time and end_time to midnight time (00:00AM UTC).
start_time = datetime.datetime(
    start_time.year, start_time.month, start_time.day, tzinfo=datetime.timezone.utc
)
end_time = datetime.datetime(
    end_time.year, end_time.month, end_time.day, tzinfo=datetime.timezone.utc
)

requested_time_range = StartManualTransferRunsRequest.TimeRange(
    start_time=start_time,
    end_time=end_time,
)

# Initialize request argument(s)
request = StartManualTransferRunsRequest(
    parent=transfer_config_name,
    requested_time_range=requested_time_range,
)

# Make the request
response = client.start_manual_transfer_runs(request=request)

# Handle the response
print("Started manual transfer runs:")
for run in response.runs:
    print(f"backfill: {run.run_time} run: {run.name}")

日志记录和监控

BigQuery Data Transfer Service 会将日志和指标导出到 Cloud Monitoring 和 Cloud Logging,以便向您的转移作业提供可观测性。您可以使用 Monitoring 设置信息中心,以监控转移作业、评估转移作业运行性能以及查看错误消息以排查转移失败问题。您可以使用 Logging 查看与转移作业运行或转移作业配置相关的日志。

此外,您还可以查看可用于 BigQuery Data Transfer Service 的审核日志,以便了解转移作业活动和数据访问日志。