Criar um tópico de importação do Amazon Kinesis Data Streams

Um tópico de importação do Amazon Kinesis Data Streams permite ingerir dados continuamente do Amazon Kinesis Data Streams como uma fonte externa e no Pub/Sub. Em seguida, é possível transmitir os dados para qualquer um dos destinos compatíveis com o Pub/Sub.

Para mais informações sobre tópicos de importação, consulte Sobre tópicos de importação.

Antes de começar

• Saiba mais sobre o processo de publicação do Pub/Sub.

• Configure as funções e permissões necessárias para gerenciar tópicos de importação do Amazon Kinesis Data Streams, incluindo o seguinte:

  • Adicione o papel de editor do Pub/Sub à conta de serviço do Pub/Sub

  • Adicionar o função do usuário da conta de serviço à conta de serviço

• Configure a federação de identidade da carga de trabalho para que Google Cloud possa acessar o serviço de streaming externo.

Papéis e permissões necessárias

Para receber as permissões necessárias para criar e gerenciar tópicos de importação do Amazon Kinesis Data Streams, peça ao administrador para conceder a você o papel do IAM Editor do Pub/Sub (roles/pubsub.editor) no seu tópico ou projeto. Para mais informações sobre a concessão de papéis, consulte Gerenciar o acesso a projetos, pastas e organizações.

Esse papel predefinido contém as permissões necessárias para criar e gerenciar tópicos de importação do Amazon Kinesis Data Streams. Para conferir as permissões exatas necessárias, expanda a seção Permissões necessárias:

Essas permissões também podem ser concedidas com funções personalizadas ou outros papéis predefinidos.

É possível configurar o controle de acesso no nível do projeto e no nível do recurso individual.

Configurar a identidade federada para acessar o Amazon Kinesis Data Streams

A federação de identidade da carga de trabalho permite que os serviços do Google Cloud acessem cargas de trabalho executadas fora do Google Cloud. Com a federação de identidade, não é necessário manter ou transmitir credenciais para Google Cloud acessar seus recursos em outras nuvens. Em vez disso, é possível usar as identidades das próprias cargas de trabalho para autenticar no Google Cloud e acessar recursos.

Criar uma conta de serviço em Google Cloud

Essa é uma etapa opcional. Se você já tiver uma conta de serviço, use-a neste procedimento em vez de criar uma nova. Se você estiver usando uma conta de serviço, acesse Registrar o ID exclusivo da conta de serviço para a próxima etapa.

Para tópicos de importação do Amazon Kinesis Data Streams, o Pub/Sub usa a conta de serviço como a identidade para acessar recursos da AWS.

Para mais informações sobre como criar uma conta de serviço, incluindo pré-requisitos, papéis e permissões obrigatórios e diretrizes de nomenclatura, consulte Criar contas de serviço. Depois de criar uma conta de serviço, talvez seja necessário aguardar 60 segundos ou mais para usá-la. Esse comportamento ocorre porque as operações de leitura têm consistência eventual e pode levar algum tempo para a nova conta de serviço ficar visível.

Registre o ID exclusivo da conta de serviço

Você precisa de um ID exclusivo da conta de serviço para configurar uma função na AWS.

1. No console Google Cloud , acesse a página de detalhes da conta de serviço.

   Acessar contas de serviço

2. Clique na conta de serviço que você acabou de criar ou na que você planeja usar.

3. Na página Detalhes da conta de serviço, anote o número do ID exclusivo.

   Você precisa do ID como parte do fluxo de trabalho para configurar uma função na AWS.

Adicionar o papel de criador de token da conta de serviço à conta de serviço do Pub/Sub

O papel de criador de token da conta de serviço (roles/iam.serviceAccountTokenCreator) permite que os principais criem credenciais de curta duração para uma conta de serviço. Esses tokens ou credenciais são usados para representar a conta de serviço.

Para mais informações sobre representação da conta de serviço, consulte Representação da conta de serviço.

Você também pode adicionar o papel de editor do Pub/Sub (roles/pubsub.publisher) durante esse procedimento. Para mais informações sobre a função e por que você está adicionando ela, consulte Adicionar a função de editor do Pub/Sub à conta de serviço do Pub/Sub.

1. No console Google Cloud , acesse a página IAM.

   Acessar IAM

2. Clique na caixa de seleção Incluir concessões de papel fornecidas pelo Google.

3. Procure a conta de serviço que tem o formato service-{PROJECT_NUMBER}@gcp-sa-pubsub.iam.gserviceaccount.com.

4. Na conta de serviço, clique no botão Editar principal.

5. Se necessário, clique em Adicionar outro papel.

6. Pesquise e clique no papel Criador de token da conta de serviço (roles/iam.serviceAccountTokenCreator).

7. Clique em Salvar.

Criar uma política na AWS

Você precisa de uma política na AWS para permitir que o Pub/Sub se autentique na AWS e ingira dados do Amazon Kinesis Data Streams.

• Para mais métodos e informações sobre como criar uma política na AWS, consulte Como criar políticas do IAM.

Para criar uma política na AWS, siga estas etapas:

1. Faça login no AWS Management Console e abra o console do IAM.

2. No painel de navegação do console do IAM, clique em Gerenciamento de acesso > Políticas.

3. Clique em Criar política.

4. Em Selecionar um serviço, clique em Kinesis.

5. Em Ação permitida, clique no seguinte:

   • List > ListShards.

     Essa ação concede permissão para listar os fragmentos em um fluxo e fornece informações sobre cada fragmento.

   • Read > SubscribeToShard.

     Essa ação concede permissão para ouvir um fragmento específico com fan-out avançado.

   • Ler > DescribeStreamConsumer.

     Essa ação concede permissão para receber a descrição de um consumidor de stream registrado.

   Essas permissões abrangem a leitura do fluxo. O Pub/Sub só aceita leitura de um stream do Kinesis com Enhanced Fan-Out usando a API SubscribeToShard de streaming.

6. Para Recursos, se você quiser restringir a política a um fluxo ou consumidor específico (recomendado), especifique o ARN do consumidor e o ARN do fluxo.

7. Clique em Adicionar mais permissões.

8. Em Selecionar um serviço, clique em STS.

9. Em Ação permitida, clique em Gravar > AssumeRoleWithWebIdentity.

   Essa ação concede permissão para obter um conjunto de credenciais de segurança temporárias para que o Pub/Sub se autentique no Amazon Kinesis Data Streams usando a federação de identidades.

10. Clique em Próxima.

11. Digite um nome e uma descrição para a política.

12. Clique em Criar política.

Criar uma função na AWS usando uma política de confiança personalizada

Você precisa criar uma função na AWS para que o Pub/Sub possa se autenticar na AWS e ingerir dados do Amazon Kinesis Data Streams.

1. Faça login no AWS Management Console e abra o console do IAM.

2. No painel de navegação do console do IAM, clique em Papéis.

3. Clique em Criar papel.

4. Em Selecionar entidade confiável, clique em Política de confiança personalizada.

5. Na seção Política de confiança personalizada, insira ou cole o seguinte:

   {
     "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>"
            }
         }
       }
     ]
   }
   

   Substitua <SERVICE_ACCOUNT_UNIQUE_ID> pelo ID exclusivo da conta de serviço que você registrou em Registrar o ID exclusivo da conta de serviço.

6. Clique em Próxima.

7. Em Adicionar permissões, pesquise e clique na política personalizada que você acabou de criar.

8. Clique em Próxima.

9. Digite um nome e uma descrição para a função.

10. Clique em Criar papel.

Adicione o papel de editor do Pub/Sub ao principal do Pub/Sub.

Para ativar a publicação, atribua um papel de editor à conta de serviço do Pub/Sub para que ele possa publicar no tópico de importação do Amazon Kinesis Data Streams.

Ativar a publicação de todos os tópicos

Use esse método se você não tiver criado nenhum tópico de importação do Amazon Kinesis Data Streams.

1. No console Google Cloud , acesse a página IAM.

   Acessar IAM

2. Clique na caixa de seleção Incluir concessões de papel fornecidas pelo Google.

3. Procure a conta de serviço que tem o formato service-{PROJECT_NUMBER}@gcp-sa-pubsub.iam.gserviceaccount.com.

4. Na conta de serviço, clique no botão Editar principal.

5. Se necessário, clique em Adicionar outro papel.

6. Pesquise e clique no papel de editor do Pub/Sub (roles/pubsub.publisher).

7. Clique em Salvar.

Ativar a publicação de um único tópico

Use esse método somente se o tópico de importação do Amazon Kinesis Data Streams já existir.

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. Execute o 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"

   Substitua:

   • TOPIC_ID: o ID do tópico de importação do Amazon Kinesis Data Streams.

   • PROJECT_NUMBER: o número do projeto. Para conferir o número do projeto, consulte Identificar projetos.

Adicione o função do usuário da conta de serviço à conta de serviço

O papel Usuário da conta de serviço (roles/iam.serviceAccountUser) inclui a permissão iam.serviceAccounts.actAs, que permite que um principal anexe uma conta de serviço às configurações de ingestão do tópico de importação do Amazon Kinesis Data Streams e use essa conta de serviço para identidade federada.

1. No console Google Cloud , acesse a página IAM.

   Acessar IAM

2. No principal que está emitindo as chamadas para criar ou atualizar tópicos, clique no botão Editar principal.

3. Se necessário, clique em Adicionar outro papel.

4. Pesquise e clique no papel de usuário da conta de serviço (roles/iam.serviceAccountUser).

5. Clique em Salvar.

Usar tópicos do Amazon Kinesis Data Streams

Você pode criar ou editar um tópico de importação.

Considerações

• Criar o tópico e a assinatura separadamente, mesmo que em rápida sucessão, pode levar à perda de dados. Há um período curto em que o tópico existe sem uma assinatura. Se algum dado for enviado ao tópico durante esse período, ele será perdido. Ao criar primeiro o tópico, depois a assinatura e, por fim, converter o tópico em um tópico de importação, você garante que nenhuma mensagem seja perdida durante o processo de importação.

Criar um tópico de importação do Amazon Kinesis Data Streams

Para saber mais sobre as propriedades associadas a um tópico, consulte Propriedades de um tópico.

Verifique se você concluiu os seguintes procedimentos:

• Configurar a identidade federada para acessar o Kinesis Data Streams

• Adicione o papel de editor do Pub/Sub à conta de serviço do Pub/Sub

• Adicionar o função do usuário da conta de serviço à conta de serviço

Para criar um tópico de importação do Amazon Kinesis Data Streams, siga estas etapas:

import (
	"context"
	"fmt"
	"io"

	pubsub "cloud.google.com/go/pubsub/v2"
	"cloud.google.com/go/pubsub/v2/apiv1/pubsubpb"
)

func createTopicWithKinesisIngestion(w io.Writer, projectID, topic string) error {
	// projectID := "my-project-id"
	// topicID := "projects/my-project-id/topics/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()

	topicpb := &pubsubpb.Topic{
		Name: topic,
		IngestionDataSourceSettings: &pubsubpb.IngestionDataSourceSettings{
			Source: &pubsubpb.IngestionDataSourceSettings_AwsKinesis_{
				AwsKinesis: &pubsubpb.IngestionDataSourceSettings_AwsKinesis{
					StreamArn:         streamARN,
					ConsumerArn:       consumerARN,
					AwsRoleArn:        awsRoleARN,
					GcpServiceAccount: gcpServiceAccount,
				},
			},
		},
	}
	topicpb, err = client.TopicAdminClient.CreateTopic(ctx, topicpb)
	if err != nil {
		return fmt.Errorf("failed to create topic with kinesis: %w", err)
	}
	fmt.Fprintf(w, "Kinesis topic created: %v\n", topicpb)
	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 mais informações sobre ARNs, consulte Nomes de recursos da Amazon (ARNs) e Identificadores do IAM.

Se você tiver problemas, consulte Resolver problemas em um tópico de importação do Amazon Kinesis Data Streams.

Editar um tópico de importação do Amazon Kinesis Data Streams

É possível editar as configurações da fonte de dados de ingestão de um tópico de importação do Amazon Kinesis Data Streams. Siga as etapas abaixo:

Cotas e limites para tópicos de importação do Amazon Kinesis Data Streams

A taxa de transferência do editor para tópicos de importação é limitada pela cota de publicação do tópico. Para mais informações, consulte Cotas e limites do Pub/Sub.

A seguir

• Monitorar um tópico de importação.

• Escolha o tipo de assinatura para seu tópico.

• Saiba como publicar uma mensagem em um tópico.

• Crie ou modifique um tópico com a CLI gcloud, as APIs REST ou as bibliotecas de cliente.

• Resolver problemas de um tópico de importação do Amazon Kinesis Data Streams.

• Etapas para configurar um job de transferência de armazenamento e transferir dados do AWS S3 para o Cloud Storage.