Créer des abonnements BigQuery

Ce document explique comment créer un abonnement BigQuery. Vous pouvez créer un abonnement BigQuery à l'aide de la console Google Cloud, de la Google Cloud CLI, de la bibliothèque cliente ou de l'API Pub/Sub.

Avant de commencer

Avant de lire ce document, assurez-vous de connaître les éléments suivants:

• Fonctionnement des abonnements

• Workflow des abonnements BigQuery

• Configurer un sujet de lettres mortes pour gérer les échecs de messages

En plus de connaître Pub/Sub et BigQuery, assurez-vous de remplir les conditions préalables suivantes avant de créer un abonnement BigQuery:

• Une table BigQuery existe. Vous pouvez également en créer un lorsque vous créez l'abonnement BigQuery, comme décrit dans les sections suivantes de ce document.

• Compatibilité entre le schéma du sujet Pub/Sub et la table BigQuery. Si vous ajoutez une table BigQuery non compatible, un message d'erreur lié à la compatibilité s'affiche. Pour en savoir plus, consultez la section Compatibilité des schémas.

Rôles et autorisations requis

Vous trouverez ci-dessous une liste de consignes concernant les rôles et les autorisations:

• Pour créer un abonnement, vous devez configurer le contrôle des accès au niveau du projet.

• Vous avez également besoin d'autorisations au niveau des ressources si vos abonnements et vos sujets se trouvent dans différents projets, comme indiqué plus loin dans cette section.

• Pour créer un abonnement BigQuery, le compte de service Pub/Sub doit être autorisé à écrire dans la table BigQuery spécifique. Pour en savoir plus sur l'octroi de ces autorisations, consultez la section suivante de ce document.

• Vous pouvez configurer un abonnement BigQuery dans un projet pour écrire dans une table BigQuery d'un autre projet.

Pour obtenir les autorisations nécessaires pour créer des abonnements BigQuery, demandez à votre administrateur de vous accorder le rôle IAM Éditeur Pub/Sub (roles/pubsub.editor) sur le projet. Pour en savoir plus sur l'attribution de rôles, consultez la page Gérer l'accès aux projets, aux dossiers et aux organisations.

Ce rôle prédéfini contient les autorisations requises pour créer des abonnements BigQuery. Pour connaître les autorisations exactes requises, développez la section Autorisations requises :

Vous pouvez également obtenir ces autorisations avec des rôles personnalisés ou d'autres rôles prédéfinis.

Si vous devez créer des abonnements BigQuery dans un projet associés à un sujet dans un autre projet, demandez à l'administrateur du sujet de vous accorder également le rôle IAM (roles/pubsub.editor) Éditeur Pub/Sub sur le sujet.

Attribuer des rôles BigQuery au compte de service Pub/Sub

Certains services Google Cloud disposent de comptes de service gérés par Google Cloud qui leur permettent d'accéder à vos ressources. Ces comptes de service sont appelés agents de service. Pub/Sub crée et gère un compte de service pour chaque projet au format service-project-number@gcp-sa-pubsub.iam.gserviceaccount.com.

Pour créer un abonnement BigQuery, le compte de service Pub/Sub doit être autorisé à écrire dans la table BigQuery spécifique et à lire ses métadonnées.

Attribuez le rôle Éditeur de données BigQuery (roles/bigquery.dataEditor) au compte de service Pub/Sub.

1. Dans la console Google Cloud, accédez à la page IAM.

   Accéder à IAM

2. Cliquez sur Accorder l'accès.

3. Dans la section Ajouter des comptes principaux, saisissez le nom de votre compte de service Pub/Sub. Le format du compte de service est service-project-number@gcp-sa-pubsub.iam.gserviceaccount.com. Par exemple, pour un projet avec project-number=112233445566, le compte de service est au format service-112233445566@gcp-sa-pubsub.iam.gserviceaccount.com.

4. Dans la section Attribuer des rôles, cliquez sur Ajouter un autre rôle.

5. Dans le menu déroulant Sélectionner un rôle, saisissez BigQuery, puis sélectionnez le rôle Éditeur de données BigQuery.

6. Cliquez sur Enregistrer.

Pour en savoir plus sur IAM BigQuery, consultez la page Rôles et autorisations BigQuery.

Propriétés des abonnements BigQuery

Lorsque vous configurez un abonnement BigQuery, vous pouvez spécifier les propriétés suivantes.

Propriétés communes

Découvrez les propriétés d'abonnement courantes que vous pouvez définir pour tous les abonnements.

Utiliser un schéma avec sujet

Cette option permet à Pub/Sub d'utiliser le schéma du sujet Pub/Sub auquel l'abonnement est associé. De plus, Pub/Sub écrit les champs des messages dans les colonnes correspondantes de la table BigQuery.

Lorsque vous utilisez cette option, n'oubliez pas de vérifier les exigences supplémentaires suivantes:

• Les champs du schéma du sujet et du schéma BigQuery doivent porter le même nom et leurs types doivent être compatibles.

• Tout champ facultatif du schéma du sujet doit également être facultatif dans le schéma BigQuery.

• Les champs obligatoires dans le schéma du sujet ne doivent pas nécessairement l'être dans le schéma BigQuery.

• Si des champs BigQuery ne figurent pas dans le schéma du sujet, ils doivent être en mode NULLABLE.

• Si le schéma du sujet comporte des champs supplémentaires qui ne sont pas présents dans le schéma BigQuery et que ces champs peuvent être supprimés, sélectionnez l'option Supprimer les champs inconnus.

• Vous ne pouvez sélectionner qu'une seule des propriétés d'abonnement, Utiliser le schéma de sujet ou Utiliser le schéma de table.

Si vous ne sélectionnez pas l'option Utiliser un schéma de sujet ou Utiliser un schéma de table, assurez-vous que la table BigQuery comporte une colonne appelée data de type BYTES, STRING ou JSON. Pub/Sub écrit le message dans cette colonne BigQuery.

Les modifications apportées au schéma des sujets Pub/Sub ou au schéma de la table BigQuery peuvent ne pas prendre effet immédiatement avec les messages écrits dans la table BigQuery. Par exemple, si l'option Supprimer les champs inconnus est activée et qu'un champ est présent dans le schéma Pub/Sub, mais pas dans le schéma BigQuery, les messages écrits dans la table BigQuery peuvent toujours ne pas contenir le champ après l'avoir ajouté au schéma BigQuery. À terme, les schémas se synchronisent et les messages suivants incluent le champ.

Lorsque vous utilisez l'option Utiliser un schéma de sujet pour votre abonnement BigQuery, vous pouvez également profiter de la capture des données modifiées (CDC, Change Data Capture) de BigQuery. La capture des données modifiées met à jour vos tables BigQuery en traitant et en appliquant les modifications aux lignes existantes.

Pour en savoir plus sur cette fonctionnalité, consultez Diffuser les mises à jour des tables avec la capture des données modifiées.

Pour savoir comment utiliser cette fonctionnalité avec des abonnements BigQuery, consultez la section Capture de données modifiées BigQuery.

Utiliser un schéma de table

Cette option permet à Pub/Sub d'utiliser le schéma de la table BigQuery pour écrire les champs d'un message JSON dans les colonnes correspondantes. Lorsque vous utilisez cette option, n'oubliez pas de vérifier les exigences supplémentaires suivantes:

• Les messages publiés doivent être au format JSON.

• Les conversions JSON suivantes sont acceptées:

  Type JSON	Type de données BigQuery
  string	NUMERIC, BIGNUMERIC, DATE, TIME, DATETIME ou TIMESTAMP
  number	NUMERIC, BIGNUMERIC, DATE, TIME, DATETIME ou TIMESTAMP

  • Lorsque vous utilisez des conversions number vers DATE, DATETIME, TIME ou TIMESTAMP, le nombre doit respecter les représentations acceptées.
  • Lorsque vous effectuez une conversion number vers NUMERIC ou BIGNUMERIC, la précision et la plage de valeurs sont limitées à celles acceptées par la norme IEEE 754 pour l'arithmétique à virgule flottante. Si vous avez besoin d'une précision élevée ou d'une plage de valeurs plus large, utilisez plutôt des conversions string vers NUMERIC ou BIGNUMERIC.
  • Lorsque vous effectuez des conversions string vers NUMERIC ou BIGNUMERIC, Pub/Sub suppose que la chaîne est un nombre lisible par l'humain (par exemple, "123.124"). Si le traitement de la chaîne en tant que nombre lisible par l'humain échoue, Pub/Sub traite la chaîne comme des octets encodés avec BigDecimalByteStringEncoder.

• Si le sujet de l'abonnement est associé à un schéma, la propriété d'encodage des messages doit être définie sur JSON.

• Si des champs BigQuery ne sont pas présents dans les messages, ils doivent être en mode NULLABLE.

• Si les messages comportent des champs supplémentaires qui ne figurent pas dans le schéma BigQuery et que ces champs peuvent être supprimés, sélectionnez l'option Supprimer les champs inconnus.

• Vous ne pouvez sélectionner qu'une seule des propriétés d'abonnement, Utiliser le schéma de sujet ou Utiliser le schéma de table.

Si vous ne sélectionnez pas l'option Utiliser un schéma de sujet ou Utiliser un schéma de table, assurez-vous que la table BigQuery comporte une colonne appelée data de type BYTES, STRING ou JSON. Pub/Sub écrit le message dans cette colonne BigQuery.

Les modifications apportées au schéma de la table BigQuery ne sont pas toujours appliquées immédiatement avec les messages écrits dans la table BigQuery. Par exemple, si l'option Supprimer les champs inconnus est activée et qu'un champ est présent dans les messages, mais pas dans le schéma BigQuery, les messages écrits dans la table BigQuery peuvent toujours ne pas contenir le champ après l'avoir ajouté au schéma BigQuery. À terme, le schéma se synchronise et les messages suivants incluent le champ.

Lorsque vous utilisez l'option Utiliser le schéma de table pour votre abonnement BigQuery, vous pouvez également profiter de la capture des données modifiées (CDC, Change Data Capture) de BigQuery. La capture des données modifiées met à jour vos tables BigQuery en traitant et en appliquant les modifications aux lignes existantes.

Pour en savoir plus sur cette fonctionnalité, consultez Diffuser les mises à jour des tables avec la capture des données modifiées.

Pour savoir comment utiliser cette fonctionnalité avec des abonnements BigQuery, consultez Capture de données modifiées BigQuery.

Supprimer les champs inconnus

Cette option est utilisée avec l'option Utiliser un schéma avec sujet ou Utiliser un schéma de table. Cette option permet à Pub/Sub de supprimer tous les champs présents dans le schéma ou le message du sujet, mais pas dans le schéma BigQuery. Si l'option Supprimer les champs inconnus n'est pas définie, les messages contenant des champs supplémentaires ne sont pas écrits dans BigQuery et restent dans la file d'attente des abonnements. L'abonnement se retrouve dans un état d'erreur.

Écrire des métadonnées

Cette option permet à Pub/Sub d'écrire les métadonnées de chaque message dans des colonnes supplémentaires de la table BigQuery. Sinon, les métadonnées ne sont pas écrites dans la table BigQuery.

Si vous sélectionnez l'option Écrire les métadonnées, assurez-vous que la table BigQuery comporte les champs décrits dans le tableau suivant.

Si vous ne sélectionnez pas l'option Écrire les métadonnées, la table BigQuery de destination ne nécessite que le champ data, sauf si use_topic_schema est défini sur "true". Si vous sélectionnez les options Écrire les métadonnées et Utiliser le schéma du sujet, le schéma du sujet ne doit contenir aucun champ dont le nom correspond à celui des paramètres de métadonnées. Cette limitation inclut les versions CamelCase de ces paramètres en casse snake.

Créer un abonnement BigQuery

Les exemples suivants montrent comment créer un abonnement avec une distribution 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

Surveiller un abonnement BigQuery

Cloud Monitoring fournit un certain nombre de métriques pour surveiller les abonnements.

Pour obtenir la liste de toutes les métriques disponibles liées à Pub/Sub et leurs descriptions, consultez la documentation de surveillance de Pub/Sub.

Vous pouvez également surveiller les abonnements à partir de Pub/Sub.

Étape suivante

• Créez ou modifiez un abonnement à l'aide des commandes gcloud.
• Créez ou modifiez un abonnement à l'aide des API REST.
• Résolvez les problèmes liés à un abonnement BigQuery.