Crea un tema de importación de Amazon Kinesis Data Streams

Un tema de importación de Amazon Kinesis Data Streams te permite transferir datos de forma continua desde Amazon Kinesis Data Streams como una fuente externa a Pub/Sub. Luego, puedes transmitir los datos a cualquiera de los destinos que admite Pub/Sub.

En este documento, se muestra cómo crear y administrar temas de importación de Amazon Kinesis Data Streams. Para crear un tema estándar, consulta Cómo crear un tema estándar.

Antes de comenzar

• 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 de Amazon Kinesis Data Streams, incluidos los siguientes:

  • Cómo agregar el rol de publicador de Pub/Sub a la cuenta de servicio de Pub/Sub

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

• Configura la federación de identidades de cargas de trabajo para que Google Cloud pueda acceder al servicio de transmisión externo.

Roles y permisos necesarios para administrar temas de importación

Para obtener los permisos que necesitas para crear y administrar temas de importación de flujos de datos de Amazon Kinesis, 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 Administra el acceso a proyectos, carpetas y organizaciones.

Este rol predefinido contiene los permisos necesarios para crear y administrar temas de importación de flujos de datos de Amazon Kinesis. 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.

Puedes configurar el control de acceso a nivel del proyecto y a nivel de los recursos individuales.

Configura la identidad federada para acceder a Kinesis Data Streams

La federación de Workload Identity permite que los servicios de Google Cloud accedan a 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 ver el paso siguiente.

En el caso de un tema de importación de AWS Kinesis Data Streams, Pub/Sub usa la cuenta de servicio como la identidad para acceder a los recursos de AWS.

Para obtener más información sobre cómo crear una cuenta de servicio, incluidos los requisitos previos, los roles y los permisos necesarios, y los lineamientos de nombres, consulta Cómo crear 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 sucede porque las operaciones de lectura tienen coherencia eventual. Es posible que la nueva cuenta de servicio tarde un poco más en ser visible.

Registra el ID único de la cuenta de servicio

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

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

   Ir a la cuenta de servicio

2. Haz clic en la cuenta de servicio que acabas de crear o en la que planeas usar.

3. En la página Detalles de la cuenta de servicio, registra el número de ID único.

   Necesitarás el ID como parte de la sección Crea un rol en AWS con una política de confianza personalizada.

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

El rol de creador de tokens de cuenta de servicio (roles/iam.serviceAccountTokenCreator) permite que las principales creen credenciales de corta duración para una cuenta de servicio. Estos tokens o credenciales se usan para suplantar la identidad 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 el rol y por qué lo agregas, consulta Cómo agregar el rol de publicador de Pub/Sub a la cuenta de servicio de Pub/Sub.

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

   Ir a IAM

2. Habilita la opción Incluir asignaciones de roles proporcionadas por Google.

3. Busca la cuenta de servicio que tiene el formato service-{PROJECT_NUMBER}@gcp-sa-pubsub.iam.gserviceaccount.com.

4. En esta cuenta de servicio, haz clic en el botón Editar principal.

5. Si es necesario, haz clic en Agregar otro rol.

6. Busca y selecciona el rol de creador de tokens de cuenta de servicio (roles/iam.serviceAccountTokenCreator).

7. 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 de modo que pueda transferir datos de un flujo 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. Te recomendamos esta práctica para que puedas restringir los permisos de 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.

• Para 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 métodos y detalles 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:

1. Accede a la consola de administración de AWS y abre la consola de IAM.

2. En el panel de navegación de la consola de IAM, haz clic en Administración de acceso > Políticas.

3. Haz clic en Crear política.

4. En Selecciona un servicio, selecciona Kinesis.

5. En acción permitida, selecciona lo siguiente:

   • List > ListShards.

     Esta acción otorga permiso para enumerar los fragmentos de un flujo y proporciona información sobre cada uno de ellos.

   • Leer > SubscribeToShard.

     Esta acción otorga permiso para escuchar un fragmento específico con fan-out mejorado.

   • Leer > DescribeStreamConsumer.

     Esta acción otorga permiso para obtener la descripción de un consumidor de transmisiones registrado.

   Estos permisos abarcan la lectura del flujo. Pub/Sub solo admite la lectura de un flujo de Kinesis con fan-out mejorado mediante la API de SubscribeToShard de transmisión.

6. En Recursos, si deseas restringir la política a un flujo o consumidor específico (recomendado), especifica el ARN del consumidor y el ARN del flujo.

7. Haz clic en Agregar más permisos.

8. En Selecciona un servicio, ingresa y selecciona STS.

9. 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.

10. Haz clic en Siguiente.

11. Ingresa un nombre y una descripción para la política.

12. Haz clic en Crear política.

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

Debes crear un rol en AWS para que Pub/Sub pueda autenticarse en AWS y transferir datos de Kinesis Data Streams.

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

1. Accede a la consola de administración de AWS y abre la consola de IAM.

2. En el panel de navegación de la consola de IAM, haz clic en Roles.

3. Haz clic en Crear rol.

4. En Seleccionar entidad de confianza, selecciona Política de confianza personalizada.

5. 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 Cómo registrar el ID único de la cuenta de servicio.

6. Haz clic en Siguiente.

7. En Agregar permisos, busca y selecciona la política personalizada que acabas de crear.

8. Haz clic en Siguiente.

9. Ingresa un nombre y una descripción para el rol.

10. Haz clic en Crear rol.

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

Debes asignar un rol de publicador a la cuenta de servicio de Pub/Sub para que Pub/Sub pueda publicar en el tema de importación desde los flujos de datos de AWS Kinesis.

• Para habilitar la publicación desde todos los temas de un proyecto, consulta Habilita la publicación desde todos los temas. Usa este método si no creaste ningún tema de importación de AWS Kinesis Data Streams.

• Para habilitar la publicación desde un tema específico (recomendado), consulta Habilita la publicación desde un solo tema. Usa este método solo si ya existe el tema de importación de AWS Kinesis Data Streams.

Habilita la publicación desde todos los temas

Usa este método si no creaste ningún tema de importación de AWS Kinesis Data Streams.

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

   Ir a IAM

2. Habilita la opción Incluir asignaciones de roles proporcionadas por Google.

3. Busca la cuenta de servicio que tiene el formato service-{PROJECT_NUMBER}@gcp-sa-pubsub.iam.gserviceaccount.com.

4. En esta cuenta de servicio, haz clic en el botón Editar principal.

5. Si es necesario, haz clic en Agregar otro rol.

6. Busca y selecciona el rol de publicador de Pub/Sub (roles/pubsub.publisher).

7. Haz clic en Guardar.

Habilita la publicación desde un solo tema

Usa este método solo si ya existe el tema de importación de AWS Kinesis Data Streams.

1. In the Google Cloud console, activate Cloud Shell.

   Activate Cloud Shell

   At the bottom of the Google Cloud console, a Cloud Shell session starts and displays a command-line prompt. Cloud Shell is a shell environment with the Google Cloud CLI already installed and with values already set for your current project. It can take a few seconds for the session to initialize.

2. 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 de Amazon Kinesis Data Streams.

   • PROJECT_NUMBER es el número del proyecto. Para ver el número del proyecto, consulta Identifica proyectos.

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

El rol de usuario de cuenta de servicio (roles/iam.serviceAccountUser) incluye el permiso iam.serviceAccounts.actAs que permite que una principal adjunte una cuenta de servicio a la configuración de transferencia del tema de importación de Amazon Kinesis Data Streams y use esa cuenta de servicio para la identidad federada.

Sigue los siguientes pasos:

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

   Ir a IAM

2. En el principal que emite las llamadas para crear o actualizar temas, haz clic en el botón Editar principal.

3. Si es necesario, haz clic en Agregar otro rol.

4. Busca y selecciona el rol de usuario de cuenta de servicio (roles/iam.serviceAccountUser).

5. Haz clic en Guardar.

Crea un tema de importación de Amazon Kinesis Data Streams

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:

• Configura la identidad federada para acceder a Kinesis Data Streams

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

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

Si creas el tema y la suscripción por separado, incluso si lo haces de forma rápida, puedes perder datos. Hay un período breve en el que el tema existe sin una suscripción. Si se envían datos al tema durante este tiempo, se perderán. Si primero creas el tema, creas la suscripción y, luego, lo conviertes en un tema de importación, garantizas que no se pierdan mensajes durante el proceso de importación.

Para crear un tema de importación de Amazon Kinesis Data Streams, sigue estos pasos:

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
}

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());
    }
  }
}

/**
 * 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 Kinesis ingestion.
  await pubSubClient.createTopic({
    name: topicNameOrId,
    ingestionDataSourceSettings: {
      awsKinesis: {
        awsRoleArn,
        gcpServiceAccount,
        streamArn,
        consumerArn,
      },
    },
  });
  console.log(`Topic ${topicNameOrId} created with AWS Kinesis ingestion.`);
}

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")

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";
}

/**
 * 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 Kinesis ingestion.
  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) y Identificadores de IAM.

Si tienes problemas, consulta Cómo solucionar problemas relacionados con un tema de importación de Amazon Kinesis Data Streams.

Cómo editar un tema de importación de Amazon Kinesis Data Streams

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

Cuotas y límites para los temas de importación de Amazon Kinesis Data Streams

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.

• Obtén información para publicar un mensaje en un tema.

• Crea o modifica un tema con gcloud CLI, APIs de REST o bibliotecas cliente.

• Cómo solucionar problemas relacionados con un tema de importación de Amazon Kinesis Data Streams

• Pasos para configurar el trabajo de transferencia de almacenamiento para transferir datos de AWS S3 a GCS.