Creare sottoscrizioni BigQuery

Questo documento descrive come creare un abbonamento a BigQuery. Puoi utilizzare la console Google Cloud, Google Cloud CLI, la libreria client o l'API Pub/Sub per creare una sottoscrizione BigQuery.

Prima di iniziare

Prima di leggere questo documento, assicurati di conoscere quanto segue:

• Come funzionano gli abbonamenti.

• Il flusso di lavoro per le sottoscrizioni BigQuery.

• Come configurare un argomento messaggi non recapitabili per gestire gli errori dei messaggi.

Oltre a conoscere Pub/Sub e BigQuery, assicurati di soddisfare i seguenti prerequisiti prima di creare una sottoscrizione BigQuery:

• Esiste una tabella BigQuery. In alternativa, puoi crearne uno quando crei l'abbonamento BigQuery come descritto nelle sezioni successive di questo documento.

• Compatibilità tra lo schema dell'argomento Pub/Sub e la tabella BigQuery. Se aggiungi una tabella BigQuery non compatibile, viene visualizzato un messaggio di errore relativo alla compatibilità. Per ulteriori informazioni, consulta la sezione Compatibilità degli schemi.

Ruoli e autorizzazioni richiesti

Di seguito è riportato un elenco di linee guida relative a ruoli e autorizzazioni:

• Per creare un abbonamento, devi configurare il controllo dell'accesso a livello di progetto.

• Sono inoltre necessarie autorizzazioni a livello di risorsa se i tuoi abbonati e gli argomenti si trovano in progetti diversi, come discusso più avanti in questa sezione.

• Per creare una sottoscrizione BigQuery, l'account di servizio Pub/Sub deve disporre dell'autorizzazione per scrivere nella tabella BigQuery specifica. Per ulteriori informazioni su come concedere queste autorizzazioni, consulta la sezione successiva di questo documento.

• Puoi configurare una sottoscrizione BigQuery in un progetto per scrivere in una tabella BigQuery di un altro progetto.

Per ottenere le autorizzazioni necessarie per creare sottoscrizioni BigQuery, chiedi all'amministratore di concederti il ruolo IAM Editor Pub/Sub (roles/pubsub.editor) nel progetto. Per saperne di più sulla concessione dei ruoli, consulta Gestire l'accesso a progetti, cartelle e organizzazioni.

Questo ruolo predefinito contiene le autorizzazioni necessarie per creare sottoscrizioni BigQuery. Per visualizzare le autorizzazioni esatte richieste, espandi la sezione Autorizzazioni richieste:

Potresti anche ottenere queste autorizzazioni con ruoli personalizzati o altri ruoli predefiniti.

Se devi creare sottoscrizioni BigQuery in un progetto associate a un argomento in un altro progetto, chiedi all'amministratore dell'argomento di concederti anche il ruolo IAM Editor Pub/Sub(roles/pubsub.editor) per l'argomento.

Assegna i ruoli BigQuery all'account di servizio Pub/Sub

Alcuni servizi Google Cloud dispongono di account di servizio gestiti da Google Cloud che consentono ai servizi di accedere alle tue risorse. Questi account di servizio sono noti come agenti di servizio. Pub/Sub crea e gestisce un account di servizio per ogni progetto nel formatoservice-project-number@gcp-sa-pubsub.iam.gserviceaccount.com.

Per creare una sottoscrizione BigQuery, l'account di servizio Pub/Sub deve disporre dell'autorizzazione per scrivere nella tabella BigQuery specifica e per leggere i metadati della tabella.

Concedi il ruolo Editor dati BigQuery (roles/bigquery.dataEditor) all'account di servizio Pub/Sub.

1. Nella console Google Cloud, vai alla pagina IAM.

   Vai a IAM

2. Fai clic su Concedi accesso.

3. Nella sezione Aggiungi entità, inserisci il nome del tuo account di servizio Pub/Sub. Il formato dell'account di servizio è service-project-number@gcp-sa-pubsub.iam.gserviceaccount.com. Ad esempio, per un progetto con project-number=112233445566, l'account di servizio è nel formato service-112233445566@gcp-sa-pubsub.iam.gserviceaccount.com.

4. Nella sezione Assegna ruoli, fai clic su Aggiungi un altro ruolo.

5. Nel menu a discesa Seleziona un ruolo, inserisci BigQuery e seleziona il ruolo Editor dati BigQuery.

6. Fai clic su Salva.

Per ulteriori informazioni su IAM di BigQuery, consulta Ruoli e autorizzazioni di BigQuery.

Proprietà delle sottoscrizioni BigQuery

Quando configuri un abbonamento BigQuery, puoi specificare le seguenti proprietà.

Proprietà comuni

Scopri le proprietà comuni degli abbonamenti che puoi impostare su tutti gli abbonamenti.

Utilizza schema argomento

Questa opzione consente a Pub/Sub di utilizzare lo schema dell'argomento Pub/Sub a cui è associata la sottoscrizione. Inoltre, Pub/Sub scrive i campi nei messaggi nelle colonne corrispondenti della tabella BigQuery.

Quando utilizzi questa opzione, ricordati di verificare i seguenti requisiti aggiuntivi:

• I campi nello schema dell'argomento e nello schema di BigQuery devono avere gli stessi nomi e i relativi tipi devono essere compatibili tra loro.

• Qualsiasi campo facoltativo nello schema dell'argomento deve essere facoltativo anche nello schema di BigQuery.

• I campi obbligatori nello schema dell'argomento non devono essere obbligatori nello schema di BigQuery.

• Se sono presenti campi BigQuery non presenti nello schema dell'argomento, questi campi devono essere in modalità NULLABLE.

• Se lo schema dell'argomento contiene campi aggiuntivi che non sono presenti nello schema di BigQuery e che possono essere ignorati, seleziona l'opzione Ignora campi sconosciuti.

• Puoi selezionare una sola delle proprietà di sottoscrizione, Utilizza lo schema dell'argomento o Utilizza lo schema della tabella.

Se non selezioni l'opzione Utilizza schema argomento o Utilizza schema tabella, assicurati che la tabella BigQuery abbia una colonna denominata data di tipo BYTES, STRING o JSON. Pub/Sub scrive il messaggio in questa colonna BigQuery.

Le modifiche allo schema degli argomenti Pub/Sub o allo schema della tabella BigQuery potrebbero non essere applicate immediatamente ai messaggi scritti nella tabella BigQuery. Ad esempio, se l'opzione Elimina campi sconosciuti è attivata e un campo è presente nello schema Pub/Sub, ma non in quello di BigQuery, i messaggi scritti nella tabella BigQuery potrebbero non contenere ancora il campo dopo averlo aggiunto allo schema di BigQuery. Alla fine, gli schemi si sincronizzano e i messaggi successivi includono il campo.

Quando utilizzi l'opzione Utilizza schema dell'argomento per la sottoscrizione BigQuery, puoi anche usufruire del CDC (Change Data Capture) di BigQuery. CDC aggiorna le tabelle BigQuery elaborando e applicando le modifiche alle righe esistenti.

Per scoprire di più su questa funzionalità, consulta Aggiornare le tabelle di streaming con l'acquisizione dei dati sulle modifiche.

Per scoprire come utilizzare questa funzionalità con gli abbonamenti BigQuery, consulta Acquisizione dei dati sulle modifiche di BigQuery.

Utilizza schema tabella

Questa opzione consente a Pub/Sub di utilizzare lo schema della tabella BigQuery per scrivere i campi di un messaggio JSON nelle colonne corrispondenti. Quando utilizzi questa opzione, ricordati di verificare i seguenti requisiti aggiuntivi:

• I messaggi pubblicati devono essere in formato JSON.

• Sono supportate le seguenti conversioni JSON:

  Tipo JSON	Tipo di dati BigQuery
  string	NUMERIC, BIGNUMERIC, DATE, TIME, DATETIME o TIMESTAMP
  number	NUMERIC, BIGNUMERIC, DATE, TIME, DATETIME o TIMESTAMP

  • Quando utilizzi le conversioni da number a DATE, DATETIME, TIME o TIMESTAMP, il numero deve rispettare le rappresentazioni supportate.
  • Quando utilizzi la conversione da number a NUMERIC o BIGNUMERIC, la precisione e l'intervallo di valori sono limitati a quelli accettati dallo standard IEEE 754 per l'aritmetica a virgola mobile. Se hai bisogno di una precisione elevata o di una gamma più ampia di valori, utilizza le conversioni da string a NUMERIC o BIGNUMERIC.
  • Quando utilizzi le conversioni da string a NUMERIC o BIGNUMERIC, Pub/Sub presuppone che la stringa sia un numero leggibile (ad es. "123.124"). Se l'elaborazione della stringa come numero leggibile non va a buon fine, Pub/Sub la tratta come byte codificati con BigDecimalByteStringEncoder.

• Se all'argomento della sottoscrizione è associato uno schema, la proprietà di codifica dei messaggi deve essere impostata su JSON.

• Se sono presenti campi BigQuery non presenti nei messaggi, questi campi BigQuery devono essere in modalità NULLABLE.

• Se i messaggi contengono campi aggiuntivi non presenti nello schema BigQuery e questi campi possono essere ignorati, seleziona l'opzione Ignora campi sconosciuti.

• Puoi selezionare una sola delle proprietà di sottoscrizione, Utilizza lo schema dell'argomento o Utilizza lo schema della tabella.

Se non selezioni l'opzione Utilizza schema argomento o Utilizza schema tabella, assicurati che la tabella BigQuery abbia una colonna denominata data di tipo BYTES, STRING o JSON. Pub/Sub scrive il messaggio in questa colonna BigQuery.

Le modifiche allo schema della tabella BigQuery potrebbero non essere applicate immediatamente ai messaggi scritti nella tabella BigQuery. Ad esempio, se l'opzione Elimina campi sconosciuti è attivata e un campo è presente nei messaggi, ma non nello schema di BigQuery, i messaggi scritti nella tabella BigQuery potrebbero non contenere ancora il campo dopo averlo aggiunto allo schema di BigQuery. Alla fine, lo schema si sincronizza e i messaggi successivi includono il campo.

Quando utilizzi l'opzione Utilizza lo schema della tabella per la tua sottoscrizione BigQuery, puoi anche sfruttare il vantaggio di BigQuery Change Data Capture (CDC). CDC aggiorna le tabelle BigQuery elaborando e applicando le modifiche alle righe esistenti.

Per scoprire di più su questa funzionalità, consulta Aggiornare le tabelle di streaming con l'acquisizione dei dati sulle modifiche.

Per scoprire come utilizzare questa funzionalità con gli abbonamenti BigQuery, consulta Acquisizione dei dati sulle modifiche di BigQuery.

Rilascia campi sconosciuti

Questa opzione viene utilizzata con l'opzione Utilizza schema argomento o Utilizza schema tabella. Questa opzione consente a Pub/Sub di ignorare qualsiasi campo presente nello schema o nel messaggio dell'argomento, ma non nello schema di BigQuery. Se non è impostato Elimina campi sconosciuti, i messaggi con campi aggiuntivi non vengono scritti in BigQuery e rimangono nel backlog della sottoscrizione. L'abbonamento termina in uno stato di errore.

Scrivi metadati

Questa opzione consente a Pub/Sub di scrivere i metadati di ogni messaggio in colonne aggiuntive della tabella BigQuery. In caso contrario, i metadati non vengono scritti nella tabella BigQuery.

Se selezioni l'opzione Scrivi metadati, assicurati che la tabella BigQuery contenga i campi descritti nella tabella seguente.

Se non selezioni l'opzione Scrivi metadati, la tabella BigQuery di destinazione richiede solo il campo data, a meno che use_topic_schema non sia true. Se selezioni sia l'opzione Scrivi metadati sia Utilizza schema argomento, lo schema dell'argomento non deve contenere campi con nomi corrispondenti a quelli dei parametri dei metadati. Questa limitazione include le versioni con lettere maiuscole di questi parametri con lettere minuscole.

Creare una sottoscrizione BigQuery

Gli esempi riportati di seguito mostrano come creare un abbonamento con invio 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

Monitorare una sottoscrizione BigQuery

Cloud Monitoring fornisce una serie di metriche per monitorare gli abbonamenti.

Per un elenco di tutte le metriche disponibili relative a Pub/Sub e le relative descrizioni, consulta la documentazione di monitoraggio per Pub/Sub.

Puoi anche monitorare le iscrizioni dall'interno di Pub/Sub.

Passaggi successivi

• Crea o modifica un abbonamento con i comandi gcloud.
• Crea o modifica un abbonamento con le API REST.
• Risolvi i problemi relativi a un abbonamento BigQuery.