Crea un argomento di importazione di Amazon Kinesis Data Streams

Un argomento di importazione di Amazon Kinesis Data Streams ti consente di importare continuamente dati da Amazon Kinesis Data Streams come origine esterna e in Pub/Sub. Poi puoi trasmettere i dati in streaming a una qualsiasi delle destinazioni supportate da Pub/Sub.

Per saperne di più sugli argomenti di importazione, consulta Informazioni sugli argomenti di importazione.

Prima di iniziare

Ruoli e autorizzazioni richiesti

Per ottenere le autorizzazioni necessarie per creare e gestire argomenti di importazione Amazon Kinesis Data Streams, chiedi all'amministratore di concederti il ruolo IAM Pub/Sub Editor (roles/pubsub.editor) nell'argomento o nel progetto. Per saperne di più sulla concessione dei ruoli, consulta Gestisci l'accesso a progetti, cartelle e organizzazioni.

Questo ruolo predefinito contiene le autorizzazioni necessarie per creare e gestire gli argomenti di importazione di Amazon Kinesis Data Streams. Per vedere quali sono esattamente le autorizzazioni richieste, espandi la sezione Autorizzazioni obbligatorie:

Autorizzazioni obbligatorie

Per creare e gestire gli argomenti di importazione di Amazon Kinesis Data Streams sono necessarie le seguenti autorizzazioni:

  • Crea un argomento di importazione: pubsub.topics.create
  • Elimina un argomento di importazione: pubsub.topics.delete
  • Recupera un argomento di importazione: pubsub.topics.get
  • Elenca un argomento di importazione: pubsub.topics.list
  • Pubblica in un argomento di importazione: pubsub.topics.publish and pubsub.serviceAgent
  • Aggiorna un argomento di importazione: pubsub.topics.update
  • Recupera il policy IAM per un argomento di importazione: pubsub.topics.getIamPolicy
  • Configura la policy IAM per un argomento di importazione: pubsub.topics.setIamPolicy

Potresti anche ottenere queste autorizzazioni con ruoli personalizzati o altri ruoli predefiniti.

Puoi configurare il controllo dell'accesso a livello di progetto e a livello di singola risorsa.

Configurare l'identità federata per accedere ad Amazon Kinesis Data Streams

La federazione delle identità per i workload consente ai servizi di accedere ai workload in esecuzione al di fuori di Google Cloud. Google Cloud Con la federazione delle identità, non devi mantenere o trasmettere le credenziali a Google Cloud per accedere alle tue risorse in altri cloud. Puoi invece utilizzare le identità dei carichi di lavoro stessi per autenticarti su Google Cloud e accedere alle risorse.

Crea un account di servizio in Google Cloud

Questo passaggio è facoltativo. Se hai già un account di servizio, puoi utilizzarlo in questa procedura anziché crearne uno nuovo. Se utilizzi un account di servizio esistente, vai a Registra l'ID univoco del service account per il passaggio successivo.

Per gli argomenti di importazione di Amazon Kinesis Data Streams, Pub/Sub utilizza l'account di servizio come identità per accedere alle risorse di AWS.

Per ulteriori informazioni sulla creazione di un account di servizio, inclusi prerequisiti, ruoli e autorizzazioni richiesti e linee guida per la denominazione, consulta Creare service account. Dopo aver creato un account di servizio, potresti dover attendere 60 secondi o più prima di account di servizio utilizzare. Questo comportamento si verifica perché le operazioni di lettura sono alla fine coerenti; potrebbe essere necessario del tempo prima che il nuovoaccount di serviziot diventi visibile.

Registra l'ID univoco del account di servizio

Per configurare un ruolo in AWS, devi disporre di un ID univoco del account di servizio.

  1. Nella Google Cloud console, vai alla pagina dei dettagli del service account.

    Vai all'account di servizio

  2. Fai clic sul account di servizio che hai appena creato o su quello che intendi utilizzare.

  3. Nella pagina Dettagli del service account, annota il numero dell'ID univoco.

    L'ID è necessario come parte del flusso di lavoro per configurare un ruolo in AWS.

Aggiungi il ruolo di creatore token account di servizio al account di servizio Pub/Sub

Il ruolo Creatore token service account (roles/iam.serviceAccountTokenCreator) consente ai principal di creare credenziali di breve durata per un account di servizio. Questi token o credenziali vengono utilizzati per simulare l'identità delaccount di serviziot.

Per ulteriori informazioni sull'impersonificazione dei account di servizio, vedi Impersonificazione dei service account.

Puoi anche aggiungere il ruolo Pub/Sub Publisher (roles/pubsub.publisher) durante questa procedura. Per saperne di più sul ruolo e sul motivo per cui lo stai aggiungendo, consulta Aggiungi il ruolo Pub/Sub Publisher al service account Pub/Sub.

  1. Nella console Google Cloud , vai alla pagina IAM.

    Vai a IAM

  2. Fai clic sulla casella di controllo Includi concessioni di ruoli fornite da Google.

  3. Cerca il account di servizio con il formato service-{PROJECT_NUMBER}@gcp-sa-pubsub.iam.gserviceaccount.com.

  4. Per questo service account, fai clic sul pulsante Modifica entità.

  5. Se necessario, fai clic su Aggiungi un altro ruolo.

  6. Cerca e fai clic sul ruolo Creatore token service account (roles/iam.serviceAccountTokenCreator).

  7. Fai clic su Salva.

Crea una policy in AWS

In AWS è necessaria una policy per consentire l'autenticazione di Pub/Sub in AWS, in modo che Pub/Sub possa importare dati da Amazon Kinesis Data Streams.

Per creare una policy in AWS, segui questi passaggi:

  1. Accedi alla Console di gestione AWS e apri la console IAM.

  2. Nel riquadro di navigazione della console per IAM, fai clic su Gestione accessi > Policy.

  3. Fai clic su Crea criterio.

  4. Per Seleziona un servizio, fai clic su Kinesis.

  5. Per Azione consentita, fai clic su:

    • List > ListShards.

      Questa azione concede l'autorizzazione a elencare gli shard in uno stream e fornisce informazioni su ciascun shard.

    • Read > SubscribeToShard.

      Questa azione concede l'autorizzazione ad ascoltare uno shard specifico con una distribuzione avanzata.

    • Leggi > DescribeStreamConsumer.

      Questa azione concede l'autorizzazione per ottenere la descrizione di un consumer di stream registrato.

    Queste autorizzazioni riguardano la lettura dallo stream. Pub/Sub supporta la lettura da un flusso Kinesis solo con Enhanced Fan-Out utilizzando l'API streaming SubscribeToShard.

  6. Per Risorse, se vuoi limitare il criterio a un flusso o a un consumatore specifico (consigliato), specifica l'ARN del consumatore e l'ARN del flusso.

  7. Fai clic su Aggiungi altre autorizzazioni.

  8. Per Seleziona un servizio, fai clic su STS.

  9. Per Azione consentita, fai clic su Scrittura > AssumeRoleWithWebIdentity.

    Questa azione concede l'autorizzazione per ottenere un insieme di credenziali di sicurezza temporanee per Pub/Sub per l'autenticazione ad Amazon Kinesis Data Streams utilizzando il federazione delle identità.

  10. Fai clic su Avanti.

  11. Inserisci un nome e una descrizione per la norma.

  12. Fai clic su Crea criterio.

Creare un ruolo in AWS utilizzando una policy di attendibilità personalizzata

Devi creare un ruolo in AWS in modo che Pub/Sub possa autenticarsi in AWS per importare i dati da Amazon Kinesis Data Streams.

  1. Accedi alla Console di gestione AWS e apri la console IAM.

  2. Nel riquadro di navigazione della console per IAM, fai clic su Ruoli.

  3. Fai clic su Crea ruolo.

  4. In Seleziona entità attendibile, fai clic su Policy di attendibilità personalizzata.

  5. Nella sezione Norma di attendibilità personalizzata, inserisci o incolla quanto segue:

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

    Sostituisci <SERVICE_ACCOUNT_UNIQUE_ID> con l'ID univoco del account di servizio che hai registrato in Registra l'ID univoco del account di servizio account.

  6. Fai clic su Avanti.

  7. Per Aggiungi autorizzazioni, cerca e fai clic sul criterio personalizzato che hai appena creato.

  8. Fai clic su Avanti.

  9. Inserisci un nome e una descrizione per il ruolo.

  10. Fai clic su Crea ruolo.

Aggiungi il ruolo Publisher Pub/Sub al principal Pub/Sub

Per attivare la pubblicazione, devi assegnare un ruolo Publisher al account di servizio Pub/Sub in modo che Pub/Sub possa pubblicare nell'argomento di importazione di Amazon Kinesis Data Streams.

Aggiungi il ruolo di agente di servizio Pub/Sub al account di servizio Pub/Sub

Per consentire a Pub/Sub di utilizzare la quota di pubblicazione del progetto dell'argomento di importazione, l'agente di servizio Pub/Sub richiede l'autorizzazione serviceusage.services.use per il progetto dell'argomento di importazione.

Per fornire questa autorizzazione, ti consigliamo di aggiungere il ruolo di agente di servizio Pub/Sub all'account di servizio Pub/Sub.

Se l'account di servizio Pub/Sub non dispone del ruolo Agente di servizio Pub/Sub, può essere concesso nel seguente modo:

  1. Nella console Google Cloud , vai alla pagina IAM.

    Vai a IAM

  2. Fai clic sulla casella di controllo Includi concessioni di ruoli fornite da Google.

  3. Cerca il account di servizio con il formato service-{PROJECT_NUMBER}@gcp-sa-pubsub.iam.gserviceaccount.com.

  4. Per questo account di servizio, fai clic sul pulsante Modifica entità.

  5. Se necessario, fai clic su Aggiungi un altro ruolo.

  6. Cerca e fai clic sul ruolo Agente di servizio Pub/Sub (roles/pubsub.serviceAgent).

  7. Fai clic su Salva.

Attivare la pubblicazione da tutti gli argomenti

Utilizza questo metodo se non hai creato argomenti di importazione di Amazon Kinesis Data Streams.

  1. Nella console Google Cloud , vai alla pagina IAM.

    Vai a IAM

  2. Fai clic sulla casella di controllo Includi concessioni di ruoli fornite da Google.

  3. Cerca il account di servizio con il formato service-{PROJECT_NUMBER}@gcp-sa-pubsub.iam.gserviceaccount.com.

  4. Per questo account di servizio, fai clic sul pulsante Modifica entità.

  5. Se necessario, fai clic su Aggiungi un altro ruolo.

  6. Cerca e fai clic sul ruolo Pub/Sub Publisher (roles/pubsub.publisher).

  7. Fai clic su Salva.

Abilita la pubblicazione da un singolo argomento

Utilizza questo metodo solo se l'argomento di importazione di Amazon Kinesis Data Streams esiste già.

  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. Esegui il 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"

    Sostituisci quanto segue:

    • TOPIC_ID: l'ID argomento dell'argomento di importazione di Amazon Kinesis Data Streams.

    • PROJECT_NUMBER: il numero di progetto. Per visualizzare il numero del progetto, consulta Identifica i progetti.

    3. Aggiungi il ruolo Utente account di servizio al account di servizio

    Il ruolo Utente service account (roles/iam.serviceAccountUser) include l'autorizzazione iam.serviceAccounts.actAs che consente a un'entità di collegare un service account alle impostazioni di importazione dell'argomento di importazione di Amazon Kinesis Data Streams e utilizzare questo account di servizio per l'identità federata.

    1. Nella console Google Cloud , vai alla pagina IAM.

      Vai a IAM

    2. Per l'entità che effettua le chiamate di creazione o aggiornamento dell'argomento, fai clic sul pulsante Modifica entità.

    3. Se necessario, fai clic su Aggiungi un altro ruolo.

    4. Cerca e fai clic sul ruolo Utente service account (roles/iam.serviceAccountUser).

    5. Fai clic su Salva.

    Utilizzare gli argomenti Amazon Kinesis Data Streams

    Puoi creare un nuovo argomento di importazione o modificarne uno esistente.

    Considerazioni

    • La creazione separata dell'argomento e della sottoscrizione, anche se eseguita in rapida successione, può comportare la perdita di dati. Esiste una breve finestra in cui l'argomento esiste senza una sottoscrizione. Se durante questo periodo vengono inviati dati all'argomento, questi vengono persi. Se crei prima l'argomento, poi la sottoscrizione e infine converti l'argomento in un argomento di importazione, garantisci che nessun messaggio venga perso durante il processo di importazione.

    Crea un argomento di importazione Amazon Kinesis Data Streams

    Per saperne di più sulle proprietà associate a un argomento, consulta Proprietà di un argomento.

    Assicurati di aver completato le seguenti procedure:

    Per creare un argomento di importazione Amazon Kinesis Data Streams:

    Console

    1. Nella console Google Cloud , vai alla pagina Argomenti.

      Vai ad Argomenti

    2. Fai clic su Crea argomento.

    3. Nel campo ID argomento, inserisci un ID per l'argomento di importazione di Amazon Kinesis Data Streams.

      Per ulteriori informazioni sulla denominazione degli argomenti, consulta le linee guida per la denominazione.

    4. Seleziona Aggiungi una sottoscrizione predefinita.

    5. Seleziona Abilita importazione.

    6. Per l'origine di importazione, seleziona Amazon Kinesis Data Streams.

    7. Inserisci i seguenti dettagli:

      • ARN del flusso Kinesis: l'ARN del flusso di dati Kinesis che prevedi di importare in Pub/Sub. Il formato ARN è il seguente: arn:${Partition}:kinesis:${Region}:${Account}:stream/${StreamName}.

      • ARN del consumer Kinesis: l'ARN della risorsa consumer registrata nel flusso di dati AWS Kinesis. Il formato ARN è il seguente: arn:${Partition}:kinesis:${Region}:${Account}:${StreamType}/${StreamName}/consumer/${ConsumerName}:${ConsumerCreationTimestamp}.

      • ARN del ruolo AWS: l'ARN del ruolo AWS. Il formato ARN del ruolo è il seguente: arn:aws:iam::${Account}:role/${RoleName}

      • Service account: il account di servizio che hai creato in Creare un account di servizio in Google Cloud.

    8. Fai clic su Crea argomento.

    gcloud

    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. Esegui il 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

      Sostituisci quanto segue:

      • TOPIC_ID: l'ID argomento.
      • KINESIS_STREAM_ARN: l'ARN di Kinesis Data Streams che prevedi di importare in Pub/Sub. Il formato dell'ARN è il seguente: arn:${Partition}:kinesis:${Region}:${Account}:stream/${StreamName}.
      • KINESIS_CONSUMER_ARN: l'ARN della risorsa consumer registrata in AWS Kinesis Data Streams. Il formato ARN è il seguente: arn:${Partition}:kinesis:${Region}:${Account}:${StreamType}/${StreamName}/consumer/${ConsumerName}:${ConsumerCreationTimestamp}.
      • KINESIS_ROLE_ARN: l'ARN del ruolo AWS. Il formato dell'ARN del ruolo è il seguente: arn:aws:iam::${Account}:role/${RoleName}.
      • PUBSUB_SERVICE_ACCOUNT: l'account di servizio che hai creato in Creare un service account in Google Cloud.

      3. C++

      Prima di provare questo esempio, segui le istruzioni di configurazione di C++ nella guida rapida all'utilizzo delle librerie client. Per saperne di più, consulta la documentazione di riferimento dell'API C++ di Pub/Sub. 

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

      Go

      L'esempio seguente utilizza la versione principale della libreria client Go Pub/Sub (v2). Se utilizzi ancora la libreria v1, consulta la guida alla migrazione alla v2. Per visualizzare un elenco di esempi di codice della versione 1, consulta gli esempi di codice deprecati.

      Prima di provare questo esempio, segui le istruzioni di configurazione di Go nella guida rapida all'utilizzo delle librerie client. Per saperne di più, consulta la documentazione di riferimento dell'API Pub/Sub Go.

      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
}

      Java

      Prima di provare questo esempio, segui le istruzioni di configurazione di Java nella guida rapida all'utilizzo delle librerie client. Per saperne di più, consulta la documentazione di riferimento dell'API Java di Pub/Sub. 

      
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

      Prima di provare questo esempio, segui le istruzioni di configurazione di Node.js nella guida rapida all'utilizzo delle librerie client. Per saperne di più, consulta la documentazione di riferimento dell'API 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 Kinesis ingestion.
  await pubSubClient.createTopic({
    name: topicNameOrId,
    ingestionDataSourceSettings: {
      awsKinesis: {
        awsRoleArn,
        gcpServiceAccount,
        streamArn,
        consumerArn,
      },
    },
  });
  console.log(`Topic ${topicNameOrId} created with AWS Kinesis ingestion.`);
}

      Node.ts

      Prima di provare questo esempio, segui le istruzioni di configurazione di Node.js nella guida rapida all'utilizzo delle librerie client. Per saperne di più, consulta la documentazione di riferimento dell'API 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
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.`);
}

      Python

      Prima di provare questo esempio, segui le istruzioni di configurazione di Python nella guida rapida all'utilizzo delle librerie client. Per saperne di più, consulta la documentazione di riferimento dell'API 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")

    Per saperne di più sugli ARN, consulta Amazon Resource Names (ARN) e Identificatori IAM.

    Se riscontri problemi, consulta l'articolo Risolvi i problemi relativi all'importazione da Amazon Kinesis Data Streams.

    Modificare un argomento di importazione Amazon Kinesis Data Streams

    Puoi modificare le impostazioni dell'origine dati di importazione di un argomento di importazione di Amazon Kinesis Data Streams. Segui questi passaggi:

    Console

    1. Nella console Google Cloud , vai alla pagina Argomenti.

      Vai ad Argomenti

    2. Fai clic sull'argomento di importazione di Amazon Kinesis Data Streams.

    3. Nella pagina dei dettagli dell'argomento, fai clic su Modifica.

    4. Aggiorna i campi che vuoi modificare.

    5. Fai clic su Aggiorna.

    gcloud

    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. Per evitare di perdere le impostazioni dell'argomento di importazione, assicurati di includerle tutte ogni volta che aggiorni l'argomento. Se ometti qualcosa, Pub/Sub reimposta l'impostazione sul valore predefinito originale.

      Esegui il comando gcloud pubsub topics update con tutti i flag menzionati nel seguente esempio:

      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

      Sostituisci quanto segue:

      • TOPIC_ID è l'ID argomento. Questo campo non può essere aggiornato.

      • KINESIS_STREAM_ARN è l'ARN di Kinesis Data Streams che prevedi di importare in Pub/Sub. Il formato dell'ARN è il seguente: arn:${Partition}:kinesis:${Region}:${Account}:stream/${StreamName}.

      • KINESIS_CONSUMER_ARN è l'ARN della risorsa consumer registrata in AWS Kinesis Data Streams. Il formato ARN è il seguente: arn:${Partition}:kinesis:${Region}:${Account}:${StreamType}/${StreamName}/consumer/${ConsumerName}:${ConsumerCreationTimestamp}.

      • KINESIS_ROLE_ARN è l'ARN del ruolo AWS. Il formato ARN del ruolo è il seguente: arn:aws:iam::${Account}:role/${RoleName}.

      • PUBSUB_SERVICE_ACCOUNT è il account di servizio che hai creato in Creare un account di servizio in Google Cloud.

    Quote e limiti per gli argomenti di importazione di Amazon Kinesis Data Streams

    Il throughput del publisher per gli argomenti di importazione è vincolato alla quota di pubblicazione dell'argomento. Per maggiori informazioni, consulta Quote e limiti di Pub/Sub.

    Passaggi successivi

    Apache Kafka® è un marchio registrato di Apache Software Foundation o delle sue affiliate negli Stati Uniti e/o in altri paesi.