Criar assinaturas do BigQuery

Neste documento, descrevemos como criar uma assinatura do BigQuery. É possível usar o console Google Cloud , a Google Cloud CLI, a biblioteca de cliente ou a API Pub/Sub para criar uma assinatura do BigQuery.

Antes de começar

Antes de ler este documento, confira se você conhece os seguintes tópicos:

• Como as assinaturas funcionam.

• O fluxo de trabalho para assinaturas do BigQuery.

• Como configurar um tópico de mensagens inativas para lidar com falhas de mensagens.

Além de conhecer o Pub/Sub e o BigQuery, verifique se você atende aos pré-requisitos a seguir antes de criar uma assinatura do BigQuery:

• Uma tabela do BigQuery existe. Outra opção é criar uma ao criar a assinatura do BigQuery, conforme descrito nas seções posteriores deste documento.

• Compatibilidade entre o esquema do tópico do Pub/Sub e a tabela do BigQuery. Se você adicionar uma tabela do BigQuery incompatível, vai receber uma mensagem de erro relacionada à compatibilidade. Para mais informações, consulte Compatibilidade de esquema.

Papéis e permissões necessárias

Confira a seguir uma lista de diretrizes sobre funções e permissões:

• Para criar uma assinatura, é necessário configurar o controle de acesso no nível do projeto.

• Você também precisa de permissões no nível do recurso se as suas assinaturas e tópicos estiverem em projetos diferentes, conforme discutido mais adiante nesta seção.

• Para criar uma assinatura do BigQuery, o agente de serviço do Pub/Sub ou uma conta de serviço personalizada precisa ter permissão para gravar na tabela específica do BigQuery. Para mais informações sobre como conceder essas permissões, consulte a próxima seção deste documento.

• É possível configurar uma assinatura do BigQuery em um projeto para gravar em uma tabela do BigQuery em outro projeto.

Para receber as permissões necessárias para criar assinaturas do BigQuery, peça ao administrador para conceder a você o papel do IAM de Editor do Pub/Sub (roles/pubsub.editor) no projeto. Para mais informações sobre a concessão de papéis, consulte Gerenciar o acesso a projetos, pastas e organizações.

Esse papel predefinido contém as permissões necessárias para criar assinaturas do BigQuery. Para conferir as permissões exatas necessárias, expanda a seção Permissões necessárias:

Essas permissões também podem ser concedidas com funções personalizadas ou outros papéis predefinidos.

Para permitir que um principal em um projeto crie uma assinatura do BigQuery em outro, conceda a esse principal o papel de editor do Pub/Sub (roles/pubsub.editor) nos dois projetos. Isso fornece as permissões necessárias para criar a nova assinatura do BigQuery e anexá-la ao tópico original. O papel de editor do Pub/Sub (roles/pubsub.editor) no tópico também ajuda a anexar assinaturas do BigQuery em um projeto diferente.

Atribuir papéis a contas de serviço

Alguns Google Cloud serviços têm contas serviço gerenciado por Google Cloudque permitem que os serviços acessem seus recursos. Essas contas de serviço são conhecidas como agentes de serviço. O Pub/Sub cria e mantém um agente de serviço para cada projeto no formato service-project-number@gcp-sa-pubsub.iam.gserviceaccount.com.

Você pode escolher entre permitir que o agente de serviço do Pub/Sub ou uma permissão de conta de serviço personalizada grave na tabela do BigQuery.

Conceder permissão ao agente de serviço do Pub/Sub significa que qualquer usuário que tenha permissão para criar uma assinatura no seu projeto pode gravar na tabela do BigQuery. Se você quiser fornecer permissões mais granulares para gravação na tabela do BigQuery, configure uma conta de serviço personalizada.

Para mais informações sobre o IAM do BigQuery, consulte Funções e permissões do BigQuery.

Atribuir papéis do BigQuery ao agente de serviço do Pub/Sub

Se você quiser criar uma assinatura do BigQuery usando o agente de serviço do Pub/Sub, ele precisa ter permissão para gravar na tabela específica do BigQuery e ler os metadados da tabela.

Conceda o papel de editor de dados do BigQuery (roles/bigquery.dataEditor) ao agente de serviço do Pub/Sub. É possível conceder a permissão em uma tabela individual ou no projeto como um todo.

Atribuir papéis do BigQuery a uma conta de serviço personalizada

Se você quiser usar uma conta de serviço personalizada para gravar em uma tabela do BigQuery, defina as seguintes permissões:

• A conta de serviço personalizada precisa ter permissão para gravar na tabela específica do BigQuery e ler os metadados dela.
• O agente de serviço do Pub/Sub precisa ter a permissão iam.serviceAccounts.getAccessToken na conta de serviço personalizada.
• O usuário que cria a assinatura precisa ter a permissão iam.serviceAccounts.actAs na conta de serviço personalizada.

Crie a conta de serviço e conceda permissões seguindo estas etapas:

1. Crie a conta de serviço personalizada. A conta de serviço precisa estar no mesmo projeto que a assinatura.

2. Conceda o papel de editor de dados do BigQuery (roles/bigquery.dataEditor) à conta de serviço personalizada.

   É possível conceder a permissão da conta de serviço em uma única tabela no projeto ou em todas as tabelas. Para fazer isso, consulte a seção apropriada em Atribuir funções do BigQuery ao agente de serviço do Pub/Sub. No procedimento, substitua o endereço de e-mail do agente de serviço do Pub/Sub pelo endereço de e-mail personalizado da conta de serviço.

3. Conceda ao agente de serviço do Pub/Sub a permissão iam.serviceAccounts.getAccessToken na conta de serviço personalizada ou em todas as contas de serviço do projeto. Para conceder essa permissão, atribua o papel roles/iam.serviceAccountTokenCreator ao agente de serviço do Pub/Sub.

   Escolha o método apropriado com base nos seus requisitos.

Se você criou a conta de serviço personalizada, já tem a permissão iam.serviceAccounts.actAs necessária. Se você precisar conceder a permissão a outra pessoa na conta de serviço:

1. No console Google Cloud , acesse a página Contas de serviço.

   Acesse as Contas de serviço

2. Insira o nome da conta de serviço personalizada no Filtro.

3. Selecione a conta de serviço na lista.

4. Clique em Pessoas com acesso.

5. Clique em Conceder acesso.

6. Na seção Adicionar principais, insira o nome da conta a que você quer conceder acesso.

7. Na lista suspensa Selecionar um papel, digite Service Account e selecione o papel Usuário da conta de serviço.

8. Além disso, se a tabela do BigQuery for do Apache Iceberg, conceda à conta de serviço do Pub/Sub o papel Administrador do Storage (roles/storage.admin) para acessar o bucket do Cloud Storage.

9. Clique em Salvar.

Propriedades de assinatura do BigQuery

Ao configurar uma assinatura do BigQuery, é possível especificar as seguintes propriedades.

Propriedades comuns

Saiba mais sobre as propriedades comuns de assinatura que podem ser definidas em todas as assinaturas.

Usar esquema de tópicos

Essa opção permite que o Pub/Sub use o esquema do tópico do Pub/Sub ao qual a assinatura está anexada. Além disso, o Pub/Sub grava os campos em mensagens nas colunas correspondentes da tabela do BigQuery.

Ao usar essa opção, verifique os seguintes requisitos adicionais:

• Os campos no esquema do tópico e no BigQuery precisam ter os mesmos nomes, e os tipos precisam ser compatíveis entre si.

• Qualquer campo opcional no esquema do tópico também precisa ser opcional no esquema do BigQuery.

• Os campos obrigatórios no esquema do tópico não precisam ser obrigatórios no esquema do BigQuery.

• Se houver campos do BigQuery que não estão presentes no esquema do tópico, eles precisam estar no modo NULLABLE.

• Se o esquema do tópico tiver campos adicionais que não estão presentes no esquema do BigQuery e que podem ser descartados, selecione a opção Drop unknown fields.

• Você pode selecionar apenas uma das propriedades de assinatura, Usar esquema de tópico ou Usar esquema de tabela.

Se você não selecionar a opção Usar esquema de tópicos ou Usar esquema de tabela, verifique se a tabela do BigQuery tem uma coluna chamada data de tipo BYTES, STRING ou JSON. O Pub/Sub grava a mensagem nessa coluna do BigQuery.

Talvez as mudanças no esquema de tópicos do Pub/Sub ou no esquema de tabela do BigQuery não entrem em vigor imediatamente com as mensagens gravadas na tabela do BigQuery. Por exemplo, se a opção Drop unknown fields estiver ativada e um campo estiver presente no esquema do Pub/Sub, mas não no BigQuery, as mensagens gravadas na tabela do BigQuery ainda poderão não conter o campo depois de adicioná-lo ao esquema do BigQuery. Eventualmente, os esquemas são sincronizados e as mensagens subsequentes incluem o campo.

Ao usar a opção Usar esquema de tópico na sua assinatura do BigQuery, você também pode aproveitar a captura de dados de mudança (CDC) do BigQuery. A CDC atualiza as tabelas do BigQuery processando e aplicando as mudanças às linhas atuais.

Para saber mais sobre esse recurso, consulte Fazer streaming de atualizações da tabela com captura de dados alterados.

Para saber como usar esse recurso com as assinaturas do BigQuery, consulte Captura de dados de alterações do BigQuery.

Usar esquema de tabela

Essa opção permite que o Pub/Sub use o esquema da tabela do BigQuery para gravar os campos de uma mensagem JSON nas colunas correspondentes. Ao usar essa opção, verifique os seguintes requisitos adicionais:

• Os nomes de cada coluna na tabela do BigQuery precisam conter apenas letras (a-z, A-Z), números (0-9) ou sublinhados (_).

• As mensagens publicadas precisam estar no formato JSON.

• As seguintes conversões JSON são compatíveis:

  Tipo de JSON	Tipo de dados do BigQuery
  string	NUMERIC, BIGNUMERIC, DATE, TIME, DATETIME ou TIMESTAMP
  number	NUMERIC, BIGNUMERIC, DATE, TIME, DATETIME ou TIMESTAMP

  • Ao usar number para conversões DATE, DATETIME, TIME ou TIMESTAMP, o número precisa seguir as representações compatíveis.
  • Ao usar a conversão de number para NUMERIC ou BIGNUMERIC, a precisão e o intervalo de valores são limitados aos aceitos pelo padrão IEEE 754 para aritmética de ponto flutuante. Se você precisar de alta precisão ou um intervalo maior de valores, use as conversões de string para NUMERIC ou BIGNUMERIC.
  • Ao usar conversões de string para NUMERIC ou BIGNUMERIC, o Pub/Sub assume que a string é um número legível por humanos (por exemplo, "123.124"). Se o processamento da string como um número legível por humanos falhar, o Pub/Sub vai tratar a string como bytes codificados com o BigDecimalByteStringEncoder.

• Se o tópico da assinatura tiver um esquema associado, a propriedade de codificação da mensagem precisará ser definida como JSON.

• Se houver campos do BigQuery que não estão presentes nas mensagens, eles precisam estar no modo NULLABLE.

• Se as mensagens tiverem campos adicionais que não estão presentes no esquema do BigQuery e esses campos puderem ser descartados, selecione a opção Drop unknown fields.

• Você pode selecionar apenas uma das propriedades de assinatura, Usar esquema de tópico ou Usar esquema de tabela.

Se você não selecionar a opção Usar esquema de tópicos ou Usar esquema de tabela, verifique se a tabela do BigQuery tem uma coluna chamada data de tipo BYTES, STRING ou JSON. O Pub/Sub grava a mensagem nessa coluna do BigQuery.

Talvez as mudanças no esquema da tabela do BigQuery não entrem em vigor imediatamente com as mensagens gravadas na tabela do BigQuery. Por exemplo, se a opção Drop unknown fields estiver ativada e um campo estiver presente nas mensagens, mas não no esquema do BigQuery, as mensagens gravadas na tabela do BigQuery ainda poderão não conter o campo depois de adicioná-lo ao esquema do BigQuery. Eventualmente, o esquema é sincronizado e as mensagens seguintes incluem o campo.

Ao usar a opção Usar esquema de tabela na sua assinatura do BigQuery, você também pode aproveitar a captura de dados alterados (CDC) do BigQuery. A CDC atualiza as tabelas do BigQuery processando e aplicando mudanças às linhas existentes.

Para saber mais sobre esse recurso, consulte Fazer streaming de atualizações da tabela com captura de dados alterados.

Para saber como usar esse recurso com as assinaturas do BigQuery, consulte Captura de dados de alterações do BigQuery.

Remover campos desconhecidos

Essa opção é usada com a opção Usar esquema de tópicos ou Usar esquema de tabela. Quando ativada, essa opção permite que o Pub/Sub descarte qualquer campo presente no esquema do tópico ou da mensagem, mas não no esquema do BigQuery. Os campos que não fazem parte do esquema do BigQuery são descartados ao gravar a mensagem na tabela do BigQuery.

Sem a opção Drop unknown fields definida, as mensagens com campos extras não são gravadas no BigQuery e permanecem no backlog das assinaturas, a menos que você configure um tópico de mensagens mortas.

A configuração Drop unknown fields não afeta campos que não são definidos no esquema de tópico do Pub/Sub ou no esquema de tabela do BigQuery. Nesse caso, uma mensagem válida do Pub/Sub é entregue à assinatura. No entanto, como o BigQuery não tem colunas definidas para esses campos extras, eles são descartados durante o processo de gravação do BigQuery. Para evitar esse comportamento, verifique se todos os campos contidos na mensagem do Pub/Sub também estão no esquema da tabela do BigQuery.

O comportamento em relação a campos extras também pode depender do tipo de esquema (Avro, buffer de protocolo) e da codificação (JSON, binário) usados. Para saber como esses fatores afetam o processamento de campos extras, consulte a documentação do seu tipo de esquema e codificação específicos.

Gravar metadados

Essa opção permite que o Pub/Sub grave os metadados de cada mensagem em outras colunas na tabela do BigQuery. Caso contrário, os metadados não serão gravados na tabela do BigQuery.

Se você selecionar a opção Gravar metadados, verifique se a tabela do BigQuery tem os campos descritos na tabela a seguir.

Se você não selecionar a opção Gravar metadados, a tabela de destino do BigQuery só vai exigir o campo data, a menos que use_topic_schema seja verdadeiro. Se você selecionar as opções Gravar metadados e Usar esquema de tópicos, o esquema do tópico não poderá conter campos com nomes que correspondam aos dos parâmetros de metadados. Essa limitação inclui versões em CamelCase desses parâmetros de caixa baixa.

Criar uma assinatura do BigQuery

Os exemplos a seguir demonstram como criar uma assinatura com entrega do BigQuery.

namespace pubsub = ::google::cloud::pubsub;
namespace pubsub_admin = ::google::cloud::pubsub_admin;
[](pubsub_admin::SubscriptionAdminClient client,
   std::string const& project_id, std::string const& topic_id,
   std::string const& subscription_id, std::string const& table_id) {
  google::pubsub::v1::Subscription request;
  request.set_name(
      pubsub::Subscription(project_id, subscription_id).FullName());
  request.set_topic(pubsub::Topic(project_id, topic_id).FullName());
  request.mutable_bigquery_config()->set_table(table_id);
  auto sub = client.CreateSubscription(request);
  if (!sub) {
    if (sub.status().code() == google::cloud::StatusCode::kAlreadyExists) {
      std::cout << "The subscription already exists\n";
      return;
    }
    throw std::move(sub).status();
  }

  std::cout << "The subscription was successfully created: "
            << sub->DebugString() << "\n";
}

using Google.Cloud.PubSub.V1;

public class CreateBigQuerySubscriptionSample
{
    public Subscription CreateBigQuerySubscription(string projectId, string topicId, string subscriptionId, string bigqueryTableId)
    {
        SubscriberServiceApiClient subscriber = SubscriberServiceApiClient.Create();
        TopicName topicName = TopicName.FromProjectTopic(projectId, topicId);
        SubscriptionName subscriptionName = SubscriptionName.FromProjectSubscription(projectId, subscriptionId);

        var subscriptionRequest = new Subscription
        {
            SubscriptionName = subscriptionName,
            TopicAsTopicName = topicName,
            BigqueryConfig = new BigQueryConfig
            {
                Table = bigqueryTableId
            }
        };
        var subscription = subscriber.CreateSubscription(subscriptionRequest);
        return subscription;
    }
}

import (
	"context"
	"fmt"
	"io"

	"cloud.google.com/go/pubsub"
)

// createBigQuerySubscription creates a Pub/Sub subscription that exports messages to BigQuery.
func createBigQuerySubscription(w io.Writer, projectID, subID string, topic *pubsub.Topic, table string) error {
	// projectID := "my-project-id"
	// subID := "my-sub"
	// topic of type https://godoc.org/cloud.google.com/go/pubsub#Topic
	// table := "my-project-id.dataset_id.table_id"
	ctx := context.Background()
	client, err := pubsub.NewClient(ctx, projectID)
	if err != nil {
		return fmt.Errorf("pubsub.NewClient: %w", err)
	}
	defer client.Close()

	sub, err := client.CreateSubscription(ctx, subID, pubsub.SubscriptionConfig{
		Topic: topic,
		BigQueryConfig: pubsub.BigQueryConfig{
			Table:         table,
			WriteMetadata: true,
		},
	})
	if err != nil {
		return fmt.Errorf("client.CreateSubscription: %w", err)
	}
	fmt.Fprintf(w, "Created BigQuery subscription: %v\n", sub)

	return nil
}

import com.google.cloud.pubsub.v1.SubscriptionAdminClient;
import com.google.pubsub.v1.BigQueryConfig;
import com.google.pubsub.v1.ProjectSubscriptionName;
import com.google.pubsub.v1.ProjectTopicName;
import com.google.pubsub.v1.Subscription;
import java.io.IOException;

public class CreateBigQuerySubscriptionExample {
  public static void main(String... args) throws Exception {
    // TODO(developer): Replace these variables before running the sample.
    String projectId = "your-project-id";
    String topicId = "your-topic-id";
    String subscriptionId = "your-subscription-id";
    String bigqueryTableId = "your-project.your-dataset.your-table";

    createBigQuerySubscription(projectId, topicId, subscriptionId, bigqueryTableId);
  }

  public static void createBigQuerySubscription(
      String projectId, String topicId, String subscriptionId, String bigqueryTableId)
      throws IOException {
    try (SubscriptionAdminClient subscriptionAdminClient = SubscriptionAdminClient.create()) {

      ProjectTopicName topicName = ProjectTopicName.of(projectId, topicId);
      ProjectSubscriptionName subscriptionName =
          ProjectSubscriptionName.of(projectId, subscriptionId);

      BigQueryConfig bigqueryConfig =
          BigQueryConfig.newBuilder().setTable(bigqueryTableId).setWriteMetadata(true).build();

      Subscription subscription =
          subscriptionAdminClient.createSubscription(
              Subscription.newBuilder()
                  .setName(subscriptionName.toString())
                  .setTopic(topicName.toString())
                  .setBigqueryConfig(bigqueryConfig)
                  .build());

      System.out.println("Created a BigQuery subscription: " + subscription.getAllFields());
    }
  }
}

/**
 * TODO(developer): Uncomment these variables before running the sample.
 */
// const topicNameOrId = 'YOUR_TOPIC_NAME_OR_ID';
// const subscriptionNameOrId = 'YOUR_SUBSCRIPTION_NAME_OR_ID';
// const bigqueryTableId = 'YOUR_TABLE_ID';

// Imports the Google Cloud client library
const {PubSub} = require('@google-cloud/pubsub');

// Creates a client; cache this for further use
const pubSubClient = new PubSub();

async function createBigQuerySubscription(
  topicNameOrId,
  subscriptionNameOrId,
  bigqueryTableId,
) {
  const options = {
    bigqueryConfig: {
      table: bigqueryTableId,
      writeMetadata: true,
    },
  };

  await pubSubClient
    .topic(topicNameOrId)
    .createSubscription(subscriptionNameOrId, options);

  console.log(`Subscription ${subscriptionNameOrId} created.`);
}

/**
 * TODO(developer): Uncomment these variables before running the sample.
 */
// const topicNameOrId = 'YOUR_TOPIC_NAME_OR_ID';
// const subscriptionNameOrId = 'YOUR_SUBSCRIPTION_NAME_OR_ID';
// const bigqueryTableId = 'YOUR_TABLE_ID';

// Imports the Google Cloud client library
import {PubSub, CreateSubscriptionOptions} from '@google-cloud/pubsub';

// Creates a client; cache this for further use
const pubSubClient = new PubSub();

async function createBigQuerySubscription(
  topicNameOrId: string,
  subscriptionNameOrId: string,
  bigqueryTableId: string,
) {
  const options: CreateSubscriptionOptions = {
    bigqueryConfig: {
      table: bigqueryTableId,
      writeMetadata: true,
    },
  };

  await pubSubClient
    .topic(topicNameOrId)
    .createSubscription(subscriptionNameOrId, options);

  console.log(`Subscription ${subscriptionNameOrId} created.`);
}

use Google\Cloud\PubSub\PubSubClient;
use Google\Cloud\PubSub\V1\BigQueryConfig;

/**
 * Creates a Pub/Sub BigQuery subscription.
 *
 * @param string $projectId  The Google project ID.
 * @param string $topicName  The Pub/Sub topic name.
 * @param string $subscriptionName  The Pub/Sub subscription name.
 * @param string $table      The BigQuery table to which to write.
 */
function create_bigquery_subscription($projectId, $topicName, $subscriptionName, $table)
{
    $pubsub = new PubSubClient([
        'projectId' => $projectId,
    ]);
    $topic = $pubsub->topic($topicName);
    $subscription = $topic->subscription($subscriptionName);
    $config = new BigQueryConfig(['table' => $table]);
    $subscription->create([
        'bigqueryConfig' => $config
    ]);

    printf('Subscription created: %s' . PHP_EOL, $subscription->name());
}

from google.cloud import pubsub_v1

# TODO(developer)
# project_id = "your-project-id"
# topic_id = "your-topic-id"
# subscription_id = "your-subscription-id"
# bigquery_table_id = "your-project.your-dataset.your-table"

publisher = pubsub_v1.PublisherClient()
subscriber = pubsub_v1.SubscriberClient()
topic_path = publisher.topic_path(project_id, topic_id)
subscription_path = subscriber.subscription_path(project_id, subscription_id)

bigquery_config = pubsub_v1.types.BigQueryConfig(
    table=bigquery_table_id, write_metadata=True
)

# Wrap the subscriber in a 'with' block to automatically call close() to
# close the underlying gRPC channel when done.
with subscriber:
    subscription = subscriber.create_subscription(
        request={
            "name": subscription_path,
            "topic": topic_path,
            "bigquery_config": bigquery_config,
        }
    )

print(f"BigQuery subscription created: {subscription}.")
print(f"Table for subscription is: {bigquery_table_id}")

require "google/cloud/pubsub"

##
# Shows how to create a BigQuery subscription where messages published
# to a topic populates a BigQuery table.
#
# @param project_id [String]
# Your Google Cloud project (e.g. "my-project")
# @param topic_id [String]
# Your topic name (e.g. "my-secret")
# @param subscription_id [String]
# ID for new subscription to be created (e.g. "my-subscription")
# @param bigquery_table_id [String]
# ID of bigquery table (e.g "my-project:dataset-id.table-id")
#
def pubsub_create_bigquery_subscription project_id:, topic_id:, subscription_id:, bigquery_table_id:
  pubsub = Google::Cloud::Pubsub.new project_id: project_id
  topic = pubsub.topic topic_id
  subscription = topic.subscribe subscription_id,
                                 bigquery_config: {
                                   table: bigquery_table_id,
                                   write_metadata: true
                                 }
  puts "BigQuery subscription created: #{subscription_id}."
  puts "Table for subscription is: #{bigquery_table_id}"
end

Monitorar uma assinatura do BigQuery

O Cloud Monitoring oferece várias métricas para monitorar assinaturas.

Para conferir uma lista de todas as métricas disponíveis relacionadas ao Pub/Sub e as descrições delas, consulte a documentação de monitoramento do Pub/Sub.

Também é possível monitorar assinaturas no Pub/Sub.

A seguir

• Crie ou modifique uma assinatura com comandos gcloud.
• Crie ou modifique uma assinatura com APIs REST.
• Resolver problemas de uma assinatura do BigQuery.