Google Ads 至 BigQuery 範本

Beta 版

這項功能 適用《服務專屬條款》。 正式發布前的功能是按照「原樣」提供，支援範圍可能有限。 詳情請參閱推出階段說明。

「Google Ads 到 BigQuery」範本屬於批次管道，可讀取 Google Ads 報表並寫入 BigQuery。

管道相關規定

• 要使用的 Google Ads 帳戶 ID。
• 用來取得資料的 Google Ads 查詢語言查詢。
• Google Ads API 的 OAuth 憑證。

範本參數

必要參數

• customerIds：用於執行查詢的 Google Ads 帳戶 ID 清單。例如：12345,67890。
• query：用於擷取資料的查詢。請參閱 Google Ads 查詢語言 (https://developers.google.com/google-ads/api/docs/query/overview)。例如：SELECT campaign.id, campaign.name FROM campaign。
• qpsPerWorker：每秒向 Google Ads 提交的查詢要求速率 (QPS)。將管道的每秒查詢數除以工作站數量上限。避免超出每個帳戶或開發人員權杖的限制。請參閱「速率限制」(https://developers.google.com/google-ads/api/docs/best-practices/rate-limits)。
• googleAdsClientId：用於識別應用程式的 OAuth 2.0 用戶端 ID。請參閱「建立用戶端 ID 和用戶端密鑰」(https://developers.google.com/google-ads/api/docs/oauth/cloud-project#create_a_client_id_and_client_secret)。
• googleAdsClientSecret：與指定用戶端 ID 相對應的 OAuth 2.0 用戶端密鑰。請參閱「建立用戶端 ID 和用戶端密鑰」(https://developers.google.com/google-ads/api/docs/oauth/cloud-project#create_a_client_id_and_client_secret)。
• googleAdsRefreshToken：用於連線至 Google Ads API 的 OAuth 2.0 更新權杖。請參閱兩步驟驗證 (https://developers.google.com/google-ads/api/docs/oauth/2sv)。
• googleAdsDeveloperToken：用於連線至 Google Ads API 的 Google Ads 開發人員權杖。請參閱「取得開發人員權杖」(https://developers.google.com/google-ads/api/docs/get-started/dev-token)。
• outputTableSpec：用於寫入輸出內容的 BigQuery 輸出資料表位置。例如 <PROJECT_ID>:<DATASET_NAME>.<TABLE_NAME>。視指定的 createDisposition 而定，系統可能會使用使用者提供的 Avro 結構定義，自動建立輸出資料表。

選用參數

• loginCustomerId：用來存取帳戶 ID 的 Google Ads 管理員帳戶 ID。例如：12345。
• bigQueryTableSchemaPath：BigQuery 結構定義 JSON 檔案的 Cloud Storage 路徑。如未設定這個值，系統會從 Proto 結構定義推斷結構定義。例如：gs://MyBucket/bq_schema.json。
• writeDisposition：BigQuery WriteDisposition (https://cloud.google.com/bigquery/docs/reference/rest/v2/Job#jobconfigurationload) 值。例如 WRITE_APPEND、WRITE_EMPTY 或 WRITE_TRUNCATE。預設值為 WRITE_APPEND。
• createDisposition：BigQuery CreateDisposition (https://cloud.google.com/bigquery/docs/reference/rest/v2/Job#jobconfigurationload)。例如 CREATE_IF_NEEDED 和 CREATE_NEVER。預設值為 CREATE_IF_NEEDED。

執行範本

控制台

1. 前往 Dataflow 的「Create job from template」(透過範本建立工作) 頁面。
前往「依範本建立工作」
2. 在「Job name」(工作名稱) 欄位中，輸入不重複的工作名稱。
3. 選用：如要使用區域端點，請從下拉式選單中選取值。預設區域為 us-central1。

   如需可執行 Dataflow 工作的地區清單，請參閱「Dataflow 位置」。

4. 從「Dataflow template」(Dataflow 範本) 下拉式選單中，選取 the Google Ads to BigQuery template。
5. 在提供的參數欄位中輸入參數值。
6. 按一下「Run Job」(執行工作)。

gcloud

注意：如要使用 Google Cloud CLI 執行 flex 範本，您必須安裝 Google Cloud CLI 版本 284.0.0 或更新版本。

在殼層或終端機中執行範本：

gcloud dataflow flex-template run JOB_NAME \
    --template-file-gcs-location=gs://dataflow-templates-REGION_NAME/VERSION/flex/Google_Ads_to_BigQuery \
    --project=PROJECT_ID \
    --region=REGION_NAME \
    --parameters \
       customerIds=CUSTOMER_IDS,\
       query=QUERY,\
       qpsPerWorker=QPS_PER_WORKER,\
       googleAdsClientId=GOOGLE_ADS_CLIENT_ID,\
       googleAdsClientSecret=GOOGLE_ADS_CLIENT_SECRET,\
       googleAdsRefreshToken=GOOGLE_ADS_REFRESH_TOKEN,\
       googleAdsDeveloperToken=GOOGLE_ADS_DEVELOPER_TOKEN,\
       outputTableSpec=OUTPUT_TABLE_SPEC,\

更改下列內容：

• JOB_NAME： 您選擇的不重複工作名稱
• VERSION：您要使用的範本版本

  您可以使用下列值：

  • latest 使用最新版範本，該範本位於值區中非依日期命名的上層資料夾： gs://dataflow-templates-REGION_NAME/latest/
  • 版本名稱，例如 2023-09-12-00_RC00，可使用特定版本的範本，該範本會以巢狀結構存放在值區中相應的依日期命名上層資料夾中：gs://dataflow-templates-REGION_NAME/
  注意：範本的最新版本可能會隨著破壞性變更而有所更新。您的實際工作環境使用的範本應該來自最近依日期命名的上層資料夾，以免這些破壞性變更影響實際運作流程。
• REGION_NAME：您要部署 Dataflow 工作的區域，例如 us-central1
• CUSTOMER_IDS：Google Ads 帳戶 ID
• QUERY：Google Ads 查詢語言查詢
• QPS_PER_WORKER：每個工作人員的必要 Google Ads 請求率
• GOOGLE_ADS_CLIENT_ID：用於識別應用程式的 OAuth 2.0 用戶端 ID
• GOOGLE_ADS_CLIENT_SECRET：與指定用戶端 ID 相對應的 OAuth 2.0 用戶端密鑰
• GOOGLE_ADS_REFRESH_TOKEN：用於連線至 Google Ads API 的 OAuth 2.0 更新權杖
• GOOGLE_ADS_DEVELOPER_TOKEN：用於連線至 Google Ads API 的 Google Ads 開發人員權杖
• OUTPUT_TABLE_SPEC：BigQuery 輸出資料表

API

如要使用 REST API 執行範本，請傳送 HTTP POST 要求。如要進一步瞭解 API 和授權範圍，請參閱 projects.templates.launch。

POST https://dataflow.googleapis.com/v1b3/projects/PROJECT_ID/locations/LOCATION/flexTemplates:launch
{
   "launchParameter": {
     "jobName": "JOB_NAME",
     "parameters": {
       "customerIds": "CUSTOMER_IDS",
       "query": "QUERY",
       "qpsPerWorker": "QPS_PER_WORKER",
       "googleAdsClientId": "GOOGLE_ADS_CLIENT_ID",
       "googleAdsClientSecret": "GOOGLE_ADS_CLIENT_SECRET",
       "googleAdsRefreshToken": "GOOGLE_ADS_REFRESH_TOKEN",
       "googleAdsDeveloperToken": "GOOGLE_ADS_DEVELOPER_TOKEN",
       "outputTableSpec": "OUTPUT_TABLE_SPEC",
     },
     "containerSpecGcsPath": "gs://dataflow-templates-LOCATION/VERSION/flex/Google_Ads_to_BigQuery",
     "environment": { "maxWorkers": "10" }
  }
}

更改下列內容：

• PROJECT_ID： 您要執行 Dataflow 工作的 Google Cloud 專案 ID
• JOB_NAME： 您選擇的不重複工作名稱
• VERSION：您要使用的範本版本

  您可以使用下列值：

  • latest 使用最新版範本，該範本位於值區中非依日期命名的上層資料夾： gs://dataflow-templates-REGION_NAME/latest/
  • 版本名稱，例如 2023-09-12-00_RC00，可使用特定版本的範本，該範本會以巢狀結構存放在值區中相應的依日期命名上層資料夾中：gs://dataflow-templates-REGION_NAME/
  注意：範本的最新版本可能會隨著破壞性變更而有所更新。您的實際工作環境使用的範本應該來自最近依日期命名的上層資料夾，以免這些破壞性變更影響實際運作流程。
• LOCATION：您要部署 Dataflow 工作的區域，例如 us-central1
• CUSTOMER_IDS：Google Ads 帳戶 ID
• QUERY：Google Ads 查詢語言查詢
• QPS_PER_WORKER：每個工作人員的必要 Google Ads 請求率
• GOOGLE_ADS_CLIENT_ID：用於識別應用程式的 OAuth 2.0 用戶端 ID
• GOOGLE_ADS_CLIENT_SECRET：與指定用戶端 ID 相對應的 OAuth 2.0 用戶端密鑰
• GOOGLE_ADS_REFRESH_TOKEN：用於連線至 Google Ads API 的 OAuth 2.0 更新權杖
• GOOGLE_ADS_DEVELOPER_TOKEN：用於連線至 Google Ads API 的 Google Ads 開發人員權杖
• OUTPUT_TABLE_SPEC：BigQuery 輸出資料表

範本原始碼

Java

/*
 * Copyright (C) 2023 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.google.cloud.teleport.v2.templates;

import com.google.ads.googleads.v19.services.GoogleAdsRow;
import com.google.cloud.teleport.metadata.Template;
import com.google.cloud.teleport.metadata.TemplateCategory;
import com.google.cloud.teleport.metadata.TemplateParameter;
import com.google.cloud.teleport.v2.options.BigQueryCommonOptions.WriteOptions;
import com.google.cloud.teleport.v2.transforms.BigQueryConverters;
import com.google.cloud.teleport.v2.transforms.GoogleAdsRowToReportRowJsonFn;
import com.google.cloud.teleport.v2.utils.GCSUtils;
import com.google.cloud.teleport.v2.utils.GoogleAdsRateLimitPolicyFactory;
import com.google.cloud.teleport.v2.utils.GoogleAdsUtils;
import com.google.common.collect.ImmutableList;
import java.util.List;
import org.apache.beam.sdk.Pipeline;
import org.apache.beam.sdk.PipelineResult;
import org.apache.beam.sdk.io.gcp.bigquery.BigQueryIO;
import org.apache.beam.sdk.io.gcp.bigquery.BigQueryIO.Write;
import org.apache.beam.sdk.io.gcp.bigquery.BigQueryIO.Write.CreateDisposition;
import org.apache.beam.sdk.io.gcp.bigquery.BigQueryIO.Write.WriteDisposition;
import org.apache.beam.sdk.io.googleads.GoogleAdsIO;
import org.apache.beam.sdk.io.googleads.GoogleAdsOptions;
import org.apache.beam.sdk.options.PipelineOptionsFactory;
import org.apache.beam.sdk.options.Validation;
import org.apache.beam.sdk.transforms.Create;
import org.apache.beam.sdk.transforms.ParDo;
import org.apache.beam.sdk.values.PCollection;
import org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.base.Strings;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/**
 * A template for writing <a href=
 * "https://developers.google.com/google-ads/api/docs/reporting/overview">Google Ads reports</a> to
 * BigQuery.
 *
 * <p>Nested fields are lifted to top-level fields by replacing the dots in field paths with
 * underscores.
 *
 * <p>Check out <a
 * href="https://github.com/GoogleCloudPlatform/DataflowTemplates/blob/main/v2/README_Google_Ads_to_BigQuery.md">README</a>
 * for instructions on how to use or modify this template.
 */
@Template(
    name = "Google_Ads_to_BigQuery",
    category = TemplateCategory.BATCH,
    displayName = "Google Ads to BigQuery",
    description =
        "The Google Ads to BigQuery template is a batch pipeline that reads Google Ads reports and writes to BigQuery.",
    optionsClass = GoogleAdsToBigQuery.GoogleAdsToBigQueryOptions.class,
    flexContainerName = "google-ads-to-bigquery",
    contactInformation = "https://cloud.google.com/support",
    documentation =
        "https://cloud.google.com/dataflow/docs/guides/templates/provided/google-ads-to-bigquery",
    requirements = {
      "The Google Ads account IDs to use.",
      "The Google Ads Query Language query to obtain information.",
      "OAuth credentials for the Google Ads API."
    },
    preview = true)
public final class GoogleAdsToBigQuery {
  public interface GoogleAdsToBigQueryOptions extends WriteOptions, GoogleAdsOptions {
    @TemplateParameter.Long(
        order = 1,
        optional = true,
        description = "Google Ads manager account ID",
        helpText = "A Google Ads manager account ID to use to access the account IDs.",
        example = "12345")
    Long getLoginCustomerId();

    void setLoginCustomerId(Long loginCustomerId);

    @TemplateParameter.Text(
        order = 2,
        regexes = {"^[0-9]+(,[0-9]+)*$"},
        description = "Google Ads account IDs",
        helpText = "A list of Google Ads account IDs to use to execute the query.",
        example = "12345,67890")
    @Validation.Required
    List<Long> getCustomerIds();

    void setCustomerIds(List<Long> customerIds);

    @TemplateParameter.Text(
        order = 3,
        description = "Google Ads Query Language query",
        helpText =
            "The query to use to get the data. See Google Ads Query Language (https://developers.google.com/google-ads/api/docs/query/overview).",
        example = "SELECT campaign.id, campaign.name FROM campaign")
    @Validation.Required
    String getQuery();

    void setQuery(String query);

    @TemplateParameter.Double(
        order = 4,
        description = "Required Google Ads request rate per worker",
        helpText =
            "The rate of query requests per second (QPS) to submit to Google Ads.  "
                + "Divide the desired per pipeline QPS by the maximum number of workers. "
                + "Avoid exceeding per-account or developer token limits. "
                + "See Rate Limits (https://developers.google.com/google-ads/api/docs/best-practices/rate-limits).")
    Double getQpsPerWorker();

    void setQpsPerWorker(Double qpsPerWorker);

    @TemplateParameter.GcsReadFile(
        order = 5,
        optional = true,
        description = "BigQuery Table Schema Path",
        helpText =
            "The Cloud Storage path to the BigQuery schema JSON file. "
                + "If this value is not set, then the schema is inferred "
                + "from the Proto schema.",
        example = "gs://MyBucket/bq_schema.json")
    String getBigQueryTableSchemaPath();

    void setBigQueryTableSchemaPath(String value);

    @TemplateParameter.Text(
        order = 6,
        description = "OAuth 2.0 Client ID identifying the application",
        helpText =
            "The OAuth 2.0 client ID that identifies the application. See Create a client ID and client secret (https://developers.google.com/google-ads/api/docs/oauth/cloud-project#create_a_client_id_and_client_secret).")
    String getGoogleAdsClientId();

    void setGoogleAdsClientId(String clientId);

    @TemplateParameter.Password(
        order = 7,
        groupName = "Source",
        description = "OAuth 2.0 Client Secret for the specified Client ID",
        helpText =
            "The OAuth 2.0 client secret that corresponds to the specified client ID. See Create a client ID and client secret (https://developers.google.com/google-ads/api/docs/oauth/cloud-project#create_a_client_id_and_client_secret).")
    String getGoogleAdsClientSecret();

    void setGoogleAdsClientSecret(String clientSecret);

    @TemplateParameter.Password(
        order = 8,
        description = "OAuth 2.0 Refresh Token for the user connecting to the Google Ads API",
        helpText =
            "The OAuth 2.0 refresh token to use to connect to the Google Ads API. See 2-Step Verification (https://developers.google.com/google-ads/api/docs/oauth/2sv).")
    String getGoogleAdsRefreshToken();

    void setGoogleAdsRefreshToken(String refreshToken);

    @TemplateParameter.Password(
        order = 9,
        description = "Google Ads developer token for the user connecting to the Google Ads API",
        helpText =
            "The Google Ads developer token to use to connect to the Google Ads API. See Obtain a developer token (https://developers.google.com/google-ads/api/docs/get-started/dev-token).")
    String getGoogleAdsDeveloperToken();

    void setGoogleAdsDeveloperToken(String developerToken);
  }

  private static final Logger LOG = LoggerFactory.getLogger(GoogleAdsToBigQuery.class);

  public static void main(String[] args) {
    run(
        PipelineOptionsFactory.fromArgs(args)
            .withValidation()
            .as(GoogleAdsToBigQueryOptions.class));
  }

  public static PipelineResult run(GoogleAdsToBigQueryOptions options) {
    Pipeline pipeline = Pipeline.create(options);
    double qps = options.getQpsPerWorker();
    String query = options.getQuery();

    PCollection<GoogleAdsRow> googleAdsRows =
        pipeline
            .apply(
                Create.of(
                    options.getCustomerIds().stream()
                        .map(Object::toString)
                        .collect(ImmutableList.toImmutableList())))
            .apply(
                GoogleAdsIO.current()
                    .read()
                    .withDeveloperToken(options.getGoogleAdsDeveloperToken())
                    .withLoginCustomerId(options.getLoginCustomerId())
                    .withQuery(options.getQuery())
                    .withRateLimitPolicy(new GoogleAdsRateLimitPolicyFactory(qps)));

    PCollection<String> reportRows =
        googleAdsRows.apply(ParDo.of(new GoogleAdsRowToReportRowJsonFn(query)));

    Write<String> write =
        BigQueryIO.<String>write()
            .withoutValidation()
            .withWriteDisposition(WriteDisposition.valueOf(options.getWriteDisposition()))
            .withCreateDisposition(CreateDisposition.valueOf(options.getCreateDisposition()))
            .withFormatFunction(BigQueryConverters::convertJsonToTableRow)
            .to(options.getOutputTableSpec());

    String schemaPath = options.getBigQueryTableSchemaPath();

    if (Strings.isNullOrEmpty(schemaPath)) {
      write = write.withSchema(GoogleAdsUtils.createBigQuerySchema(query));
    } else {
      write = write.withJsonSchema(GCSUtils.getGcsFileAsString(schemaPath));
    }

    reportRows.apply("WriteToBigQuery", write);

    return pipeline.run();
  }
}

後續步驟

• 瞭解 Dataflow 範本。
• 請參閱 Google 提供的範本清單。