Crea suscripciones a BigQuery

En este documento, se describe cómo crear una suscripción a BigQuery. Puedes usar la consola de Google Cloud , Google Cloud CLI, la biblioteca cliente o la API de Pub/Sub para crear una suscripción a BigQuery.

Antes de comenzar

Antes de leer este documento, asegúrate de estar familiarizado con lo siguiente:

• Cómo funcionan las suscripciones

• El flujo de trabajo de las suscripciones a BigQuery

• Cómo configurar un tema de mensajes no entregados para controlar las fallas de los mensajes

Además de conocer Pub/Sub y BigQuery, asegúrate de cumplir con los siguientes requisitos previos antes de crear una suscripción a BigQuery:

• Que exista una tabla de BigQuery Como alternativa, puedes crear uno cuando crees la suscripción a BigQuery, como se describe en las secciones posteriores de este documento.

• La compatibilidad entre el esquema del tema de Pub/Sub y la tabla de BigQuery. Si agregas una tabla de BigQuery que no es compatible, recibirás un mensaje de error relacionado con la compatibilidad. Para obtener más información, consulta Compatibilidad de esquemas.

Roles y permisos requeridos

A continuación, se muestra una lista de lineamientos relacionados con los roles y los permisos:

• Para crear una suscripción, debes configurar el control de acceso a nivel del proyecto.

• También necesitas permisos a nivel del recurso si tus suscripciones y temas están en diferentes proyectos, como se explica más adelante en esta sección.

• Para crear una suscripción de BigQuery, el agente de servicio de Pub/Sub o una cuenta de servicio personalizada deben tener permiso para escribir en la tabla de BigQuery específica. Para obtener más información sobre cómo otorgar estos permisos, consulta la siguiente sección de este documento.

• Puedes configurar una suscripción a BigQuery en un proyecto para escribir en una tabla de BigQuery en otro proyecto.

Para obtener los permisos que necesitas para crear suscripciones de BigQuery, pídele a tu administrador que te otorgue el rol de IAM de Editor de Pub/Sub (roles/pubsub.editor) en el proyecto. Para obtener más información sobre cómo otorgar roles, consulta Administra el acceso a proyectos, carpetas y organizaciones.

Este rol predefinido contiene los permisos necesarios para crear suscripciones de BigQuery. Para ver los permisos exactos que son necesarios, expande la sección Permisos requeridos:

También puedes obtener estos permisos con roles personalizados o con otros roles predefinidos.

Para permitir que un principal de un proyecto cree una suscripción a BigQuery en otro proyecto, debes otorgarle el rol de editor de Pub/Sub (roles/pubsub.editor) en ambos proyectos. Esto proporciona los permisos necesarios para crear la nueva suscripción a BigQuery y adjuntarla al tema original. El rol de editor de Pub/Sub (roles/pubsub.editor) en el tema también te ayuda a adjuntar suscripciones de BigQuery en un proyecto diferente al tema.

Asigna roles a cuentas de servicio

Algunos Google Cloud servicios tienen Google Cloudcuentas de servicio administradas que permiten que los servicios accedan a tus recursos. Estas cuentas de servicio se conocen como agentes de servicio. Pub/Sub crea y mantiene un agente de servicio para cada proyecto en el formato service-project-number@gcp-sa-pubsub.iam.gserviceaccount.com.

Puedes elegir entre permitir que el agente de servicio de Pub/Sub o una cuenta de servicio personalizada escriban en la tabla de BigQuery.

Otorgar permiso al agente de servicio de Pub/Sub significa que cualquier usuario que tenga permiso para crear una suscripción en tu proyecto puede escribir en la tabla de BigQuery. Si deseas proporcionar un permiso más detallado para escribir en la tabla de BigQuery, configura una cuenta de servicio personalizada.

Para obtener más información sobre IAM de BigQuery, consulta Roles y permisos de BigQuery.

Asigna roles de BigQuery al agente de servicio de Pub/Sub

Si deseas crear una suscripción a BigQuery con el agente de servicio de Pub/Sub, este debe tener permiso para escribir en la tabla de BigQuery específica y leer sus metadatos.

Otorga el rol de editor de datos de BigQuery (roles/bigquery.dataEditor) al agente de servicio de Pub/Sub. Puedes otorgar el permiso en una tabla individual o en todo el proyecto.

Asigna roles de BigQuery a una cuenta de servicio personalizada

Si deseas usar una cuenta de servicio personalizada para escribir en una tabla de BigQuery, debes establecer los siguientes permisos:

• La cuenta de servicio personalizada debe tener permiso para escribir en la tabla de BigQuery específica y para leer los metadatos de la tabla.
• El agente de servicio de Pub/Sub debe tener el permiso iam.serviceAccounts.getAccessToken en la cuenta de servicio personalizada.
• El usuario que crea la suscripción debe tener el permiso iam.serviceAccounts.actAs en la cuenta de servicio personalizada.

Crea la cuenta de servicio y otorga permisos con los siguientes pasos:

1. Crea la cuenta de servicio personalizada. La cuenta de servicio debe estar en el mismo proyecto que la suscripción.

2. Otorga el rol de editor de datos de BigQuery (roles/bigquery.dataEditor) a la cuenta de servicio personalizada.

   Puedes otorgar el permiso de cuenta de servicio en una sola tabla del proyecto o en todas las tablas del proyecto. Para ello, consulta la sección correspondiente en Asigna roles de BigQuery al agente de servicio de Pub/Sub. En el procedimiento, reemplaza la dirección de correo electrónico del agente de servicio de Pub/Sub por la dirección de correo electrónico personalizada de la cuenta de servicio.

3. Otórgale al agente de servicio de Pub/Sub el permiso iam.serviceAccounts.getAccessToken en la cuenta de servicio personalizada o en todas las cuentas de servicio del proyecto. Para otorgar este permiso, asigna el rol roles/iam.serviceAccountTokenCreator al agente de servicio de Pub/Sub.

   Elige el método adecuado según tus requisitos.

Si creaste la cuenta de servicio personalizada, ya deberías tener el permiso iam.serviceAccounts.actAs necesario. Si necesitas otorgarle a otra persona el permiso en la cuenta de servicio, haz lo siguiente:

1. En la consola de Google Cloud , ve a la página Cuentas de servicio.

   Ir a Cuentas de servicio

2. Ingresa el nombre de la cuenta de servicio personalizada en Filtro.

3. Selecciona la cuenta de servicio de la lista.

4. Haz clic en Principales con acceso.

5. Haz clic en Otorgar acceso.

6. En la sección Agregar principales, ingresa el nombre de la cuenta a la que deseas otorgar acceso.

7. En el menú desplegable Seleccionar un rol, ingresa Service Account y selecciona el rol Usuario de cuenta de servicio.

8. Además, si tu tabla de BigQuery es una tabla de Apache Iceberg, otorga a la cuenta de servicio de Pub/Sub el rol de administrador de almacenamiento (roles/storage.admin) para acceder al bucket de Cloud Storage.

9. Haz clic en Guardar.

Propiedades de suscripción a BigQuery

Cuando configuras una suscripción a BigQuery, puedes especificar las siguientes propiedades.

Propiedades comunes

Obtén información sobre las propiedades de suscripción comunes que puedes configurar en todas las suscripciones.

Usar el esquema de tema

Esta opción permite que Pub/Sub use el esquema del tema de Pub/Sub al que está adjunta la suscripción. Además, Pub/Sub escribe los campos de los mensajes en las columnas correspondientes de la tabla de BigQuery.

Cuando uses esta opción, recuerda verificar los siguientes requisitos adicionales:

• Los campos del esquema del tema y del esquema de BigQuery deben tener los mismos nombres, y sus tipos deben ser compatibles entre sí.

• Cualquier campo opcional en el esquema de tema también debe ser opcional en el esquema de BigQuery.

• Los campos obligatorios en el esquema del tema no es necesario que sean obligatorios en el esquema de BigQuery.

• Si hay campos de BigQuery que no están presentes en el esquema del tema, estos campos de BigQuery deben estar en el modo NULLABLE.

• Si el esquema de tema tiene campos adicionales que no están presentes en el esquema de BigQuery y estos campos se pueden descartar, selecciona la opción Descartar campos desconocidos.

• Puedes seleccionar solo una de las propiedades de suscripción, Usar esquema de tema o Usar esquema de tabla.

Si no seleccionas la opción Usar el esquema de tema o Usar el esquema de tabla, asegúrate de que la tabla de BigQuery tenga una columna llamada data de tipo BYTES, STRING o JSON. Pub/Sub escribe el mensaje en esta columna de BigQuery.

Es posible que no veas que los cambios en el esquema de temas de Pub/Sub o el esquema de tablas de BigQuery se apliquen de inmediato con los mensajes que se escriben en la tabla de BigQuery. Por ejemplo, si la opción Descartar campos desconocidos está habilitada y hay un campo presente en el esquema de Pub/Sub, pero no en el esquema de BigQuery, es posible que los mensajes escritos en la tabla de BigQuery aún no contengan el campo después de agregarlo al esquema de BigQuery. Con el tiempo, los esquemas se sincronizan y los mensajes posteriores incluyen el campo.

Cuando usas la opción Usar esquema de tema para tu suscripción a BigQuery, también puedes aprovechar la captura de datos modificados (CDC) de BigQuery. La CDC actualiza las tablas de BigQuery mediante el procesamiento y la aplicación de cambios a las filas existentes.

Para obtener más información sobre esta función, consulta Transmite actualizaciones de tablas con captura de datos modificados.

Para obtener información sobre cómo usar esta función con suscripciones a BigQuery, consulta Captura de datos modificados de BigQuery.

Usar el esquema de tabla

Esta opción permite que Pub/Sub use el esquema de la tabla de BigQuery para escribir los campos de un mensaje en JSON en las columnas correspondientes. Cuando uses esta opción, recuerda verificar los siguientes requisitos adicionales:

• Los nombres de cada columna de la tabla de BigQuery solo deben contener letras (a-z, A-Z), números (0-9) o guiones bajos (_).

• Los mensajes publicados deben estar en formato JSON.

• Se admiten las siguientes conversiones de JSON:

  Tipo de JSON	Tipo de datos de BigQuery
  string	NUMERIC, BIGNUMERIC, DATE, TIME, DATETIME o TIMESTAMP
  number	NUMERIC, BIGNUMERIC, DATE, TIME, DATETIME o TIMESTAMP

  • Cuando se usan conversiones de number a DATE, DATETIME, TIME o TIMESTAMP, el número debe cumplir con las representaciones admitidas.
  • Cuando se usa la conversión de number a NUMERIC o BIGNUMERIC, la precisión y el rango de valores se limitan a los que acepta el estándar IEEE 754 para aritmética de punto flotante. Si necesitas una precisión alta o un rango más amplio de valores, usa las conversiones de string a NUMERIC o BIGNUMERIC.
  • Cuando se usan conversiones de string a NUMERIC o BIGNUMERIC, Pub/Sub supone que la cadena es un número legible por humanos (p.ej., "123.124"). Si el procesamiento de la cadena como un número legible por humanos falla, Pub/Sub la trata como bytes codificados con BigDecimalByteStringEncoder.

• Si el tema de la suscripción tiene un esquema asociado, la propiedad de codificación de mensajes debe establecerse en JSON.

• Si hay campos de BigQuery que no están presentes en los mensajes, estos campos de BigQuery deben estar en el modo NULLABLE.

• Si los mensajes tienen campos adicionales que no están presentes en el esquema de BigQuery y estos campos se pueden descartar, selecciona la opción Descartar campos desconocidos.

• Puedes seleccionar solo una de las propiedades de suscripción, Usar esquema de tema o Usar esquema de tabla.

Si no seleccionas la opción Usar el esquema de tema o Usar el esquema de tabla, asegúrate de que la tabla de BigQuery tenga una columna llamada data de tipo BYTES, STRING o JSON. Pub/Sub escribe el mensaje en esta columna de BigQuery.

Es posible que no veas los cambios en el esquema de la tabla de BigQuery de inmediato con los mensajes escritos en la tabla de BigQuery. Por ejemplo, si la opción Descartar campos desconocidos está habilitada y un campo está presente en los mensajes, pero no en el esquema de BigQuery, es posible que los mensajes escritos en la tabla de BigQuery aún no contengan el campo después de agregarlo al esquema de BigQuery. Con el tiempo, se sincroniza el esquema y los mensajes posteriores incluyen el campo.

Cuando usas la opción Usar esquema de tabla para tu suscripción a BigQuery, también puedes aprovechar la captura de datos modificados (CDC) de BigQuery. La CDC actualiza tus tablas de BigQuery mediante el procesamiento y la aplicación de cambios a las filas existentes.

Para obtener más información sobre esta función, consulta Transmite actualizaciones de tablas con captura de datos modificados.

Para obtener información sobre cómo usar esta función con suscripciones a BigQuery, consulta Captura de datos modificados de BigQuery.

Quitar campos desconocidos

Esta opción se usa con las opciones Usar el esquema de tema o Usar el esquema de tabla. Cuando se habilita, esta opción permite que Pub/Sub descarte cualquier campo que esté presente en el esquema o mensaje del tema, pero no en el esquema de BigQuery. Los campos que no forman parte del esquema de BigQuery se descartan cuando se escribe el mensaje en la tabla de BigQuery.

Si no se establece Descartar campos desconocidos, los mensajes con campos adicionales no se escriben en BigQuery y permanecen en las tareas pendientes de la suscripción, a menos que configures un tema de mensajes no entregados.

La configuración Descartar campos desconocidos no afecta a los campos que no están definidos en el esquema de tema de Pub/Sub ni en el esquema de tabla de BigQuery. En este caso, se entrega un mensaje de Pub/Sub válido a la suscripción. Sin embargo, debido a que BigQuery no tiene columnas definidas para estos campos adicionales, estos campos se descartan durante el proceso de escritura de BigQuery. Para evitar este comportamiento, asegúrate de que cualquier campo contenido en el mensaje de Pub/Sub también esté contenido en el esquema de la tabla de BigQuery.

El comportamiento en relación con los campos adicionales también puede depender del tipo de esquema específico (Avro, búfer de protocolo) y la codificación (JSON, binario) que se use. Para obtener información sobre cómo estos factores afectan el manejo de campos adicionales, consulta la documentación de tu tipo de esquema y codificación específicos.

Escribir metadatos

Esta opción permite que Pub/Sub escriba los metadatos de cada mensaje en columnas adicionales de la tabla de BigQuery. De lo contrario, los metadatos no se escriben en la tabla de BigQuery.

Si seleccionas la opción Escribir metadatos, asegúrate de que la tabla de BigQuery tenga los campos que se describen en la siguiente tabla.

Si no seleccionas la opción Escribir metadatos, la tabla de BigQuery de destino solo requiere el campo data, a menos que use_topic_schema sea verdadero. Si seleccionas las opciones Escribir metadatos y Usar esquema del tema, el esquema del tema no debe contener ningún campo con nombres que coincidan con los de los parámetros de metadatos. Esta limitación incluye las versiones en mayúsculas y minúsculas de estos parámetros en formato de barra invertida.

Crea una suscripción a BigQuery

En los siguientes ejemplos, se muestra cómo crear una suscripción con entrega de 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

Supervisa una suscripción a BigQuery

Cloud Monitoring proporciona varias métricas para supervisar las suscripciones.

Para obtener una lista de todas las métricas disponibles relacionadas con Pub/Sub y sus descripciones, consulta la documentación de supervisión de Pub/Sub.

También puedes supervisar las suscripciones desde dentro de Pub/Sub.

¿Qué sigue?

• Crea o modifica una suscripción con los comandos gcloud.
• Crea o modifica una suscripción con las APIs de REST.
• Soluciona problemas relacionados con una suscripción a BigQuery.