Crea un tema de importación

Un tema de importación te permite transferir datos de fuentes externas a Pub/Sub. Luego, puedes transmitir los datos a cualquiera de los destinos que admite Pub/Sub.

Pub/Sub admite Amazon Kinesis Data Streams como una fuente externa para transferir datos a un tema de importación.

Descripción general de la importación de temas

Un tema de importación tiene habilitada la transferencia de temas como una propiedad. Esto permite que un tema de importación transfiera datos de transmisión. Puedes habilitar la transferencia en un tema a través de la consola, Google Cloud CLI, las llamadas de REST o las bibliotecas cliente. Como parte de la administración del tema de importación, Google Cloud proporciona supervisión y escalamiento de la canalización de transferencia.

Sin un tema de importación, la transmisión de datos a Pub/Sub desde una fuente de datos requiere un servicio adicional. Este servicio adicional extrae datos de la fuente original y los publica en Pub/Sub. El servicio adicional puede ser un motor de transmisión, como Apache Spark, o un servicio personalizado. También debes configurar, implementar, ejecutar, escalar y supervisar este servicio.

A continuación, se incluye una lista de información importante sobre los temas de importación:

Al igual que con un tema estándar, aún puedes publicar de forma manual en un tema de importación.
Solo se puede adjuntar una única fuente de transferencia a un tema de importación.

Te recomendamos que importes temas para los datos de transmisión. Si estás considerando la transferencia de datos por lotes a BigQuery en lugar de la transferencia de datos de transmisión, puedes probar el Servicio de transferencia de datos de BigQuery (BQ DTS). Si deseas transferir datos a Cloud Storage, el Servicio de transferencia de almacenamiento (STS) es una buena opción.

Antes de comenzar

Obtén más información sobre el servicio de Pub/Sub y su terminología.
Obtén más información sobre el proceso de publicación de Pub/Sub.
Configura los roles y permisos necesarios para administrar los temas de importación, lo que incluye agregar el rol de publicador de Pub/Sub a la cuenta de servicio de Pub/Sub y agregar el rol del usuario de la cuenta de servicio a la cuenta de servicio.
Configura la federación de identidades para cargas de trabajo para que Google Cloud pueda acceder al servicio de transmisión externo.

Roles y permisos obligatorios para administrar los temas de importación

Si quieres obtener los permisos que necesitas para crear y administrar temas de importación, pídele a tu administrador que te otorgue el rol de IAM de editor de Pub/Sub(roles/pubsub.editor) en tu tema o proyecto. Para obtener más información sobre cómo otorgar roles, consulta Cómo administrar el acceso.

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

Permisos necesarios

Se requieren los siguientes permisos para crear y administrar temas de importación:

Crea un tema de importación: pubsub.topics.create
Borra un tema de importación: pubsub.topics.delete
Obtén un tema de importación: pubsub.topics.get
Genera una lista de un tema de importación: pubsub.topics.list
Publicar en un tema de importación: pubsub.topics.publish
Actualiza un tema de importación: pubsub.topics.update
Obtén la política de IAM para un tema de importación: pubsub.topics.getIamPolicy
Configura la política de IAM para un tema de importación: pubsub.topics.setIamPolicy

Es posible que también puedas obtener estos permisos con funciones personalizadas o con otras funciones predefinidas.

Puedes configurar el control de acceso a nivel de proyecto y a nivel de recurso individual.

Configura una identidad federada para acceder a Kinesis Data Streams

La federación de identidades para cargas de trabajo permite que los servicios de Google Cloud accedan a las cargas de trabajo que se ejecutan fuera de Google Cloud. Con la federación de identidades, no necesitas mantener ni pasar credenciales a Google Cloud para acceder a tus recursos en otras nubes. En su lugar, puedes usar las identidades de las cargas de trabajo para autenticarte en Google Cloud y acceder a los recursos.

Crea una cuenta de servicio en Google Cloud

Este paso es opcional. Si ya tienes una cuenta de servicio, puedes usarla en este procedimiento en lugar de crear una nueva. Si usas una cuenta de servicio existente, ve a Registra el ID único de la cuenta de servicio para realizar el siguiente paso.

Para un tema de importación, Pub/Sub usa la cuenta de servicio como la identidad a fin de acceder a los recursos de AWS.

Para obtener más información sobre cómo crear una cuenta de servicio, incluidos los requisitos previos, las funciones y los permisos necesarios, y los lineamientos para asignar nombres, consulta Crea cuentas de servicio. Después de crear una cuenta de servicio, es posible que debas esperar 60 segundos o más antes de usarla. Este comportamiento ocurre porque las operaciones de lectura tienen coherencia eventual y la cuenta de servicio nueva puede tardar en volverse visible.

Registra el ID único de la cuenta de servicio

Necesitas un ID único de cuenta de servicio para configurar un rol en la consola de AWS.

En la consola de Google Cloud, ve a la página de detalles de la Cuenta de servicio.

Ir a la cuenta de servicio
Haz clic en la cuenta de servicio que acabas de crear o en la que planeas usar.
En la página Detalles de la cuenta de servicio, registra el número de ID único.

Lo necesitarás como parte de la sección Crea un rol en AWS mediante una política de confianza personalizada.

Agrega el rol de creador de tokens de cuentas de servicio a la cuenta de servicio de Pub/Sub

La función de creador de tokens de cuenta de servicio (roles/iam.serviceAccountTokenCreator) permite a las principales crear credenciales de corta duración para una cuenta de servicio. Estos tokens o credenciales se usan para actuar en nombre de la cuenta de servicio.

Para obtener más información sobre el uso de la identidad de cuentas de servicio, consulta Uso de la identidad de cuentas de servicio.

También puedes agregar el rol de publicador de Pub/Sub (roles/pubsub.publisher) durante este procedimiento. Para obtener más información sobre la función y por qué la agregas, consulta Agrega la función de publicador de Pub/Sub a la cuenta de servicio de Pub/Sub.

En la consola de Google Cloud, ve a la página IAM.

Ir a IAM
Habilita la opción Incluir asignaciones de roles proporcionados por Google.
Busca la cuenta de servicio con el formato service-{PROJECT_NUMBER}@gcp-sa-pubsub.iam.gserviceaccount.com.
Para esta cuenta de servicio, haz clic en el botón Editar principal.
Si es necesario, haz clic en Agregar otro rol.
Busca y selecciona el rol de creador de tokens de cuentas de servicio (roles/iam.serviceAccountTokenCreator).
Haz clic en Guardar.

Crea una política en AWS

Necesitas una política en AWS para permitir que Pub/Sub se autentique con AWS para que Pub/Sub pueda transferir datos desde una transmisión de datos de AWS Kinesis. Antes de crear una política de AWS, crea un flujo de datos de Kinesis y un consumidor registrado en él. Recomendamos esta práctica para que puedas restringir los permisos a la transmisión específica.

Para obtener más información sobre cómo crear un flujo de datos de AWS Kinesis, consulta Flujo de datos de Kinesis.
Si deseas obtener más información sobre la API de flujo de datos de AWS Kinesis que se usa para registrar consumidores, consulta RegisterStreamConsumer.
Para obtener más información y métodos sobre cómo crear una política en AWS, consulta Crea políticas de IAM.

Para crear una política en AWS, sigue estos pasos:

Accede a la consola de administración de AWS y abre la consola de IAM.
En el panel de navegación de la consola para IAM, haz clic en Administración de accesos > Políticas.
Haz clic en Crear política.
En Select a service, selecciona Kinesis.
En Se permite la acción, selecciona lo siguiente:
- List > ListShards.
  
  Esta acción otorga permiso para enumerar los fragmentos de una transmisión y proporciona información sobre cada uno.
- Lee > SubscribeToShard.
  
  Esta acción otorga permiso para escuchar un fragmento específico con fan-out mejorado.
- Lee > DescribeStreamConsumer.
  
  Esta acción otorga permiso para obtener la descripción de un consumidor de transmisión registrado.
Estos permisos incluyen la lectura de contenido de la transmisión. Pub/Sub solo admite la lectura desde una transmisión de Kinesis con una fan-out mejorada mediante la API de transmisión de SubscribeToShard.
En Recursos, si deseas restringir la política a una transmisión o un consumidor específicos (recomendado), especifica el ARN del consumidor y el ARN de transmisión.
Haz clic en Agregar más permisos.
En Seleccionar un servicio, ingresa y selecciona STS.
En Acción permitida, selecciona Escribir > AssumeRoleWithWebIdentity.

Esta acción otorga permiso para obtener un conjunto de credenciales de seguridad temporales para que Pub/Sub se autentique en el flujo de datos de Kinesis mediante la federación de identidades.
Haz clic en Siguiente.
Ingresa un nombre y una descripción para la política.
Haz clic en Crear política.

Crea un rol en AWS mediante una política de confianza personalizada

Debes crear una función en AWS para que Pub/Sub pueda autenticarse en AWS y transferir datos desde Kinesis Data Streams.

Para crear un rol con una política de confianza personalizada, sigue estos pasos:

Accede a la consola de administración de AWS y abre la consola de IAM.
En el panel de navegación de la consola para IAM, haz clic en Roles.
Haz clic en Crear rol.
En Seleccionar entidad de confianza, selecciona Política de confianza personalizada.

En la sección Política de confianza personalizada, ingresa o pega lo siguiente:

{
  "Version": "2012-10-17",
  "Statement": [
    {
     "Effect": "Allow",
     "Principal": {
        "Federated": "accounts.google.com"
     },
     "Action": "sts:AssumeRoleWithWebIdentity",
     "Condition": {
         "StringEquals": {
           "accounts.google.com:sub": "<SERVICE_ACCOUNT_UNIQUE_ID>"
         }
      }
    }
  ]
}

Reemplaza <SERVICE_ACCOUNT_UNIQUE_ID> por el ID único de la cuenta de servicio que registraste en Registra el ID único de la cuenta de servicio.

Haz clic en Siguiente.
En Agregar permisos, busca y selecciona la política personalizada que acabas de crear.
Haz clic en Siguiente.
Ingresa un nombre y una descripción para el rol.
Haz clic en Crear rol.

Agrega el rol de publicador de Pub/Sub a la cuenta de servicio de Pub/Sub

Debes asignar una función de publicador a la cuenta de servicio de Pub/Sub para que Pub/Sub pueda publicar en el tema de importación desde AWS Kinesis Data Streams.

Para habilitar la publicación a partir de todos los temas de un proyecto, consulta Habilita la publicación de todos los temas.
Para habilitar la publicación a partir de un tema específico (recomendado), consulta Habilita la publicación desde un solo tema.

Habilitar la publicación de todos los temas

En la consola de Google Cloud, ve a la página IAM.

Ir a IAM
Habilita la opción Incluir asignaciones de roles proporcionados por Google.
Busca la cuenta de servicio que tenga el formato service-{PROJECT_NUMBER}@gcp-sa-pubsub.iam.gserviceaccount.com.
Para esta cuenta de servicio, haz clic en el botón Editar principal.
Si es necesario, haz clic en Agregar otro rol.
Busca y selecciona el rol de publicador de Pub/Sub (roles/pubsub.publisher).
Haz clic en Guardar.

Habilita la publicación desde un solo tema

Si quieres otorgar el permiso de publicación solo a un tema de importación específico, sigue estos pasos:

En la consola de Google Cloud, activa Cloud Shell.

Activar Cloud Shell

En la parte inferior de la consola de Google Cloud, se inicia una sesión de Cloud Shell en la que se muestra una ventana de línea de comandos. Cloud Shell es un entorno de shell con Google Cloud CLI ya instalada y con valores ya establecidos para el proyecto actual. La sesión puede tardar unos segundos en inicializarse.
Ejecuta el comando gcloud pubsub topics add-iam-policy-binding:
```
gcloud pubsub topics add-iam-policy-binding TOPIC_ID\
   --member="serviceAccount:service-PROJECT_NUMBER@gcp-sa-pubsub.iam.gserviceaccount.com"\
   --role="roles/pubsub.publisher"
```
Reemplaza lo siguiente:
- TOPIC_ID es el ID del tema de importación.
- PROJECT_NUMBER es el número del proyecto. Para ver el número de proyecto, consulta Identifica proyectos.

Agrega el rol de usuario de cuenta de servicio a la cuenta de servicio

La función de usuario de cuenta de servicio (roles/iam.serviceAccountUser) incluye el permiso iam.serviceAccounts.actAs que permite a una principal conectar una cuenta de servicio a la configuración de transferencia del tema de importación y usar esa cuenta de servicio para una identidad federada.

Sigue los siguientes pasos:

En la consola de Google Cloud, ve a la página IAM.

Ir a IAM
En la principal que emite las llamadas para crear o actualizar temas, haz clic en el botón Editar principal.
Si es necesario, haz clic en Agregar otro rol.
Busca y selecciona el rol del usuario de la cuenta de servicio (roles/iam.serviceAccountUser).
Haz clic en Guardar.

Crea un tema de importación

Para obtener más información sobre las propiedades asociadas con un tema, consulta Propiedades de un tema.

Asegúrate de haber completado los siguientes procedimientos:

Para crear un tema de importación, sigue estos pasos:

Console

En la consola de Google Cloud, ve a la página Temas.

Ir a temas
Haz clic en Crear tema.
En el campo ID del tema, ingresa un ID para tu tema de importación.

Para obtener más información sobre cómo asignar nombres a temas, consulta los lineamientos para asignar nombres.
Selecciona Agregar una suscripción predeterminada.
Selecciona Habilitar transferencia.
En la fuente de transferencia, selecciona Amazon Kinesis Data Streams.
Ingresa los siguientes detalles:
- ARN de transmisión de Kinesis: Es el ARN de la transmisión de datos de Kinesis que planeas transferir a Pub/Sub. El formato de ARN es el siguiente: arn:${Partition}:kinesis:${Region}:${Account}:stream/${StreamName}.
- ARN del consumidor de Kinesis: Es el ARN del recurso del consumidor que está registrado en el flujo de datos de Kinesis de AWS. El formato de ARN es el siguiente: arn:${Partition}:kinesis:${Region}:${Account}:${StreamType}/${StreamName}/consumer/${ConsumerName}:${ConsumerCreationTimpstamp}.
- ARN de función de AWS: Es el ARN de la función de AWS. El formato ARN de la función es el siguiente: arn:aws:iam:${Account}:role/${RoleName}.
- Cuenta de servicio: Es la cuenta de servicio que creaste en Crea una cuenta de servicio en Google Cloud.
Haz clic en Crear tema.

gcloud

En la consola de Google Cloud, activa Cloud Shell.

Activar Cloud Shell

En la parte inferior de la consola de Google Cloud, se inicia una sesión de Cloud Shell en la que se muestra una ventana de línea de comandos. Cloud Shell es un entorno de shell con Google Cloud CLI ya instalada y con valores ya establecidos para el proyecto actual. La sesión puede tardar unos segundos en inicializarse.
Ejecuta el comando gcloud pubsub topics create:
```
gcloud pubsub topics create TOPIC_ID 

       --kinesis-ingestion-stream-arn KINESIS_STREAM_ARN

       --kinesis-ingestion-consumer-arn KINESIS_CONSUMER_ARN

       --kinesis-ingestion-role-arn KINESIS_ROLE_ARN

       --kinesis-ingestion-service-account PUBSUB_SERVICE_ACCOUNT
```
Reemplaza lo siguiente:
- TOPIC_ID es el ID del tema.
- KINESIS_STREAM_ARN es el ARN de los flujos de datos de Kinesis que planeas transferir a Pub/Sub. El formato de ARN es el siguiente: arn:${Partition}:kinesis:${Region}:${Account}:stream/${StreamName}.
- KINESIS_CONSUMER_ARN es el ARN del recurso del consumidor que está registrado en Kinesis Data Streams de AWS. El formato de ARN es el siguiente: arn:${Partition}:kinesis:${Region}:${Account}:${StreamType}/${StreamName}/consumer/${ConsumerName}:${ConsumerCreationTimpstamp}.
- KINESIS_ROLE_ARN es el ARN de la función de AWS. El formato de ARN de la función es el siguiente: arn:aws:iam:${Account}:role/${RoleName}.
- PUBSUB_SERVICE_ACCOUNT es la cuenta de servicio que creaste en Crea una cuenta de servicio en Google Cloud.

Go

Antes de probar esta muestra, sigue las instrucciones de configuración de Go que se encuentran en la guía de inicio rápido de Pub/Sub con bibliotecas cliente. Si quieres obtener más información, consulta la documentación de referencia de la API de Pub/Sub Go.

Para autenticarte en Pub/Sub, configura las credenciales predeterminadas de la aplicación. Si deseas obtener más información, consulta Configura la autenticación para un entorno de desarrollo local.

import (
	"context"
	"fmt"
	"io"

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

func createTopicWithKinesisIngestion(w io.Writer, projectID, topicID string) error {
	// projectID := "my-project-id"
	// topicID := "my-topic"
	streamARN := "stream-arn"
	consumerARN := "consumer-arn"
	awsRoleARN := "aws-role-arn"
	gcpServiceAccount := "gcp-service-account"

	ctx := context.Background()
	client, err := pubsub.NewClient(ctx, projectID)
	if err != nil {
		return fmt.Errorf("pubsub.NewClient: %w", err)
	}
	defer client.Close()

	cfg := &pubsub.TopicConfig{
		IngestionDataSourceSettings: &pubsub.IngestionDataSourceSettings{
			Source: &pubsub.IngestionDataSourceAWSKinesis{
				StreamARN:         streamARN,
				ConsumerARN:       consumerARN,
				AWSRoleARN:        awsRoleARN,
				GCPServiceAccount: gcpServiceAccount,
			},
		},
	}
	t, err := client.CreateTopicWithConfig(ctx, topicID, cfg)
	if err != nil {
		return fmt.Errorf("CreateTopic: %w", err)
	}
	fmt.Fprintf(w, "Kinesis topic created: %v\n", t)
	return nil
}

Java

Antes de probar esta muestra, sigue las instrucciones de configuración de Java que se encuentran en la guía de inicio rápido de Pub/Sub con bibliotecas cliente. Si quieres obtener más información, consulta la documentación de referencia de la API de Pub/Sub Java.


import com.google.cloud.pubsub.v1.TopicAdminClient;
import com.google.pubsub.v1.IngestionDataSourceSettings;
import com.google.pubsub.v1.Topic;
import com.google.pubsub.v1.TopicName;
import java.io.IOException;

public class CreateTopicWithKinesisIngestionExample {
  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";
    // Kinesis ingestion settings.
    String streamArn = "stream-arn";
    String consumerArn = "consumer-arn";
    String awsRoleArn = "aws-role-arn";
    String gcpServiceAccount = "gcp-service-account";

    createTopicWithKinesisIngestionExample(
        projectId, topicId, streamArn, consumerArn, awsRoleArn, gcpServiceAccount);
  }

  public static void createTopicWithKinesisIngestionExample(
      String projectId,
      String topicId,
      String streamArn,
      String consumerArn,
      String awsRoleArn,
      String gcpServiceAccount)
      throws IOException {
    try (TopicAdminClient topicAdminClient = TopicAdminClient.create()) {
      TopicName topicName = TopicName.of(projectId, topicId);

      IngestionDataSourceSettings.AwsKinesis awsKinesis =
          IngestionDataSourceSettings.AwsKinesis.newBuilder()
              .setStreamArn(streamArn)
              .setConsumerArn(consumerArn)
              .setAwsRoleArn(awsRoleArn)
              .setGcpServiceAccount(gcpServiceAccount)
              .build();
      IngestionDataSourceSettings ingestionDataSourceSettings =
          IngestionDataSourceSettings.newBuilder().setAwsKinesis(awsKinesis).build();

      Topic topic =
          topicAdminClient.createTopic(
              Topic.newBuilder()
                  .setName(topicName.toString())
                  .setIngestionDataSourceSettings(ingestionDataSourceSettings)
                  .build());

      System.out.println("Created topic with Kinesis ingestion settings: " + topic.getAllFields());
    }
  }
}

Node.js

Antes de probar esta muestra, sigue las instrucciones de configuración de Node.js que se encuentran en la guía de inicio rápido de Pub/Sub con bibliotecas cliente. Si quieres obtener más información, consulta la documentación de referencia de la API de Pub/Sub Node.js.

/**
 * TODO(developer): Uncomment these variables before running the sample.
 */
// const topicNameOrId = 'YOUR_TOPIC_NAME_OR_ID';
// const roleArn = 'arn:aws:iam:...';
// const gcpServiceAccount = 'ingestion-account@...';
// const streamArn = 'arn:aws:kinesis:...';
// const consumerArn = 'arn:aws:kinesis:...';

// 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 createTopicWithKinesisIngestion(
  topicNameOrId,
  awsRoleArn,
  gcpServiceAccount,
  streamArn,
  consumerArn
) {
  // Creates a new topic with a schema. Note that you might also
  // pass Encodings.Json or Encodings.Binary here.
  await pubSubClient.createTopic({
    name: topicNameOrId,
    ingestionDataSourceSettings: {
      awsKinesis: {
        awsRoleArn,
        gcpServiceAccount,
        streamArn,
        consumerArn,
      },
    },
  });
  console.log(`Topic ${topicNameOrId} created with AWS Kinesis ingestion.`);
}

Python

Antes de probar esta muestra, sigue las instrucciones de configuración de Python que se encuentran en la guía de inicio rápido de Pub/Sub con bibliotecas cliente. Si quieres obtener más información, consulta la documentación de referencia de la API de Pub/Sub Python.

from google.cloud import pubsub_v1
from google.pubsub_v1.types import Topic
from google.pubsub_v1.types import IngestionDataSourceSettings

# TODO(developer)
# project_id = "your-project-id"
# topic_id = "your-topic-id"
# stream_arn = "your-stream-arn"
# consumer_arn = "your-consumer-arn"
# aws_role_arn = "your-aws-role-arn"
# gcp_service_account = "your-gcp-service-account"

publisher = pubsub_v1.PublisherClient()
topic_path = publisher.topic_path(project_id, topic_id)

request = Topic(
    name=topic_path,
    ingestion_data_source_settings=IngestionDataSourceSettings(
        aws_kinesis=IngestionDataSourceSettings.AwsKinesis(
            stream_arn=stream_arn,
            consumer_arn=consumer_arn,
            aws_role_arn=aws_role_arn,
            gcp_service_account=gcp_service_account,
        )
    ),
)

topic = publisher.create_topic(request=request)

print(f"Created topic: {topic.name} with AWS Kinesis Ingestion Settings")

C++

Antes de probar esta muestra, sigue las instrucciones de configuración de C++ que se encuentran en la guía de inicio rápido de Pub/Sub con bibliotecas cliente. Si quieres obtener más información, consulta la documentación de referencia de la API de Pub/Sub C++.

namespace pubsub = ::google::cloud::pubsub;
namespace pubsub_admin = ::google::cloud::pubsub_admin;
[](pubsub_admin::TopicAdminClient client, std::string project_id,
   std::string topic_id, std::string stream_arn, std::string consumer_arn,
   std::string aws_role_arn, std::string gcp_service_account) {
  google::pubsub::v1::Topic request;
  request.set_name(
      pubsub::Topic(std::move(project_id), std::move(topic_id)).FullName());
  auto* aws_kinesis =
      request.mutable_ingestion_data_source_settings()->mutable_aws_kinesis();
  aws_kinesis->set_stream_arn(stream_arn);
  aws_kinesis->set_consumer_arn(consumer_arn);
  aws_kinesis->set_aws_role_arn(aws_role_arn);
  aws_kinesis->set_gcp_service_account(gcp_service_account);

  auto topic = client.CreateTopic(request);
  // Note that kAlreadyExists is a possible error when the library retries.
  if (topic.status().code() == google::cloud::StatusCode::kAlreadyExists) {
    std::cout << "The topic already exists\n";
    return;
  }
  if (!topic) throw std::move(topic).status();

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

Node.js (TypeScript)

/**
 * TODO(developer): Uncomment these variables before running the sample.
 */
// const topicNameOrId = 'YOUR_TOPIC_NAME_OR_ID';
// const roleArn = 'arn:aws:iam:...';
// const gcpServiceAccount = 'ingestion-account@...';
// const streamArn = 'arn:aws:kinesis:...';
// const consumerArn = 'arn:aws:kinesis:...';

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

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

async function createTopicWithKinesisIngestion(
  topicNameOrId: string,
  awsRoleArn: string,
  gcpServiceAccount: string,
  streamArn: string,
  consumerArn: string
) {
  // Creates a new topic with a schema. Note that you might also
  // pass Encodings.Json or Encodings.Binary here.
  await pubSubClient.createTopic({
    name: topicNameOrId,
    ingestionDataSourceSettings: {
      awsKinesis: {
        awsRoleArn,
        gcpServiceAccount,
        streamArn,
        consumerArn,
      },
    },
  });
  console.log(`Topic ${topicNameOrId} created with AWS Kinesis ingestion.`);
}

Para obtener más información sobre los ARN, consulta Nombres de recursos de Amazon (ARN) e Identificadores de IAM.

Si tienes problemas, consulta Cómo solucionar problemas de importación.

Cómo editar un tema de importación

Puedes editar la configuración de la fuente de datos de transferencia de un tema de importación. Sigue los siguientes pasos:

Console

En la consola de Google Cloud, ve a la página Temas.

Ir a temas
Haz clic en el tema de importación.
En la página de detalles del tema, haz clic en Editar.
Actualiza los campos que quieres cambiar.
Haz clic en Update.

gcloud

En la consola de Google Cloud, activa Cloud Shell.

Activar Cloud Shell

En la parte inferior de la consola de Google Cloud, se inicia una sesión de Cloud Shell en la que se muestra una ventana de línea de comandos. Cloud Shell es un entorno de shell con Google Cloud CLI ya instalada y con valores ya establecidos para el proyecto actual. La sesión puede tardar unos segundos en inicializarse.
Ejecuta el comando gcloud pubsub topics update con todas las marcas mencionadas en el siguiente ejemplo:
```
  gcloud pubsub topics update TOPIC_ID 

         --kinesis-ingestion-stream-arn KINESIS_STREAM_ARN

         --kinesis-ingestion-consumer-arn KINESIS_CONSUMER_ARN

         --kinesis-ingestion-role-arn KINESIS_ROLE_ARN

         --kinesis-ingestion-service-account PUBSUB_SERVICE_ACCOUNT
  
```
Reemplaza lo siguiente:
- TOPIC_ID es el ID del tema. No se puede actualizar este campo.
- KINESIS_STREAM_ARN es el ARN de los flujos de datos de Kinesis que planeas transferir a Pub/Sub. El formato de ARN es el siguiente: arn:${Partition}:kinesis:${Region}:${Account}:stream/${StreamName}.
- KINESIS_CONSUMER_ARN es el ARN del recurso del consumidor que está registrado en Kinesis Data Streams de AWS. El formato de ARN es el siguiente: arn:${Partition}:kinesis:${Region}:${Account}:${StreamType}/${StreamName}/consumer/${ConsumerName}:${ConsumerCreationTimpstamp}.
- KINESIS_ROLE_ARN es el ARN de la función de AWS. El formato de ARN de la función es el siguiente: arn:aws:iam:${Account}:role/${RoleName}.
- PUBSUB_SERVICE_ACCOUNT es la cuenta de servicio que creaste en Crea una cuenta de servicio en Google Cloud.

Cuotas y límites de los temas de importación

La capacidad de procesamiento del publicador para los temas de importación está limitada por la cuota de publicación del tema. Para obtener más información, consulta Cuotas y límites de Pub/Sub.

¿Qué sigue?

Supervisa un tema de importación.
Elige el tipo de suscripción para tu tema.
Aprende a publicar un mensaje en un tema.
Crea o modifica un tema con la gcloud CLI, las APIs de REST o las bibliotecas cliente.
Solucionar problemas en un tema de importación
Pasos para configurar el trabajo de transferencia de almacenamiento para transferir datos de AWS S3 a GCS.