Creare un argomento di importazione

Un argomento di importazione ti consente di importare i dati da origini esterne in Pub/Sub. Puoi quindi trasmettere i flussi di dati in qualsiasi destinazione supportata da Pub/Sub.

Pub/Sub supporta Stream di dati Amazon Kinesis come origine esterna per l'importazione di dati in un argomento di importazione.

Importa panoramica argomento

In un argomento di importazione è abilitata l'importazione per l'argomento come proprietà. Ciò consente a un argomento di importare i flussi di dati. Puoi abilitare l'importazione per un argomento utilizzando la console, Google Cloud CLI, le chiamate REST o le librerie client. Nell'ambito della gestione dell'argomento di importazione, Google Cloud fornisce il monitoraggio e la scalabilità della pipeline di importazione.

Senza un argomento di importazione, il trasferimento di flussi di dati in Pub/Sub da un'origine dati richiede un servizio aggiuntivo. Questo servizio aggiuntivo estrae i dati dall'origine originale e li pubblica in Pub/Sub. Il servizio aggiuntivo può essere un motore di flussi di dati come Apache Spark o un servizio personalizzato. Devi inoltre configurare, eseguire il deployment, eseguire, scalare e monitorare questo servizio.

Di seguito è riportato un elenco di importanti informazioni riguardanti gli argomenti relativi all'importazione:

• Analogamente a un argomento standard, puoi comunque pubblicare manualmente in un argomento di importazione.

• Puoi collegare una sola origine di importazione a un argomento di importazione.

Ti consigliamo di importare gli argomenti per i flussi di dati. Se stai prendendo in considerazione l'importazione dati in batch in BigQuery anziché l'importazione dati, puoi provare BigQuery Data Transfer Service (BQ DTS). Se vuoi importare dati in Cloud Storage, Storage Transfer Service (STS) è una buona opzione.

Prima di iniziare

• Scopri di più sul servizio Pub/Sub e sulla sua terminologia.

• Scopri di più sul processo di pubblicazione di Pub/Sub.

• Configura i ruoli e le autorizzazioni necessari per gestire gli argomenti di importazione, ad esempio aggiungendo il ruolo di publisher Pub/Sub all'account di servizio Pub/Sub e l'aggiunta del ruolo utente dell'account di servizio all'account di servizio.

• Configura la federazione delle identità per i carichi di lavoro in modo che Google Cloud possa accedere al servizio di streaming esterno.

Autorizzazioni e ruoli richiesti per gestire gli argomenti di importazione

Per ottenere le autorizzazioni necessarie per creare e gestire gli argomenti di importazione, chiedi all'amministratore di concederti il ruolo IAM Editor Pub/Sub(roles/pubsub.editor) per il tuo argomento o progetto. Per saperne di più sulla concessione dei ruoli, consulta Gestire l'accesso.

Questo ruolo predefinito contiene le autorizzazioni necessarie per creare e gestire gli argomenti di importazione. Per visualizzare le autorizzazioni esatte necessarie, espandi la sezione Autorizzazioni richieste:

Potresti anche essere in grado di ottenere queste autorizzazioni con i ruoli personalizzati o altri ruoli predefiniti.

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

Configura l'identità federata per accedere a Kinesis Data Streams

La federazione di Workload Identity consente ai servizi Google Cloud di accedere ai carichi di lavoro in esecuzione al di fuori di Google Cloud. Con la federazione delle identità, non è necessario gestire o passare le credenziali a Google Cloud per accedere alle risorse in altri cloud. Puoi invece utilizzare le identità dei carichi di lavoro per eseguire l'autenticazione in Google Cloud e accedere alle risorse.

Creare 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 Registrare l'ID univoco dell'account di servizio per il passaggio successivo.

Per un argomento di importazione, Pub/Sub utilizza l'account di servizio come identità per accedere alle risorse da AWS.

Per maggiori informazioni sulla creazione di un account di servizio, inclusi prerequisiti, ruoli e autorizzazioni richiesti e linee guida per la denominazione, consulta la pagina Creare account di servizio. Dopo aver creato un account di servizio, potrebbe essere necessario attendere almeno 60 secondi prima di utilizzarlo. Questo comportamento si verifica perché le operazioni di lettura sono alla fine coerenti. Può essere necessario del tempo prima che il nuovo account di servizio diventi visibile.

Registra l'ID univoco dell'account di servizio

Per configurare un ruolo nella console AWS, devi avere un ID univoco dell'account di servizio.

1. Nella console Google Cloud, vai alla pagina dei dettagli dell'Account di servizio.

   Vai all'account di servizio

2. Fai clic sull'account di servizio appena creato o su quello che prevedi di utilizzare.

3. Nella pagina Dettagli account di servizio, registra il numero ID univoco.

   Ti servirà in quanto parte della sezione Crea un ruolo in AWS utilizzando un criterio di attendibilità personalizzato.

Aggiungere il ruolo Creatore token account di servizio all'account di servizio Pub/Sub

Il ruolo Creatore token account di servizio (roles/iam.serviceAccountTokenCreator) consente alle entità di creare credenziali di breve durata per un account di servizio. Questi token o credenziali vengono utilizzati per rappresentare l'account di servizio.

Per ulteriori informazioni sulla simulazione dell'identità degli account di servizio, consulta Impersonificazione degli account di servizio.

Puoi anche aggiungere il ruolo di publisher Pub/Sub (roles/pubsub.publisher) durante questa procedura. Per ulteriori informazioni sul ruolo e sul motivo per cui lo stai aggiungendo, consulta Aggiungere il ruolo di publisher Pub/Sub all'account di servizio Pub/Sub.

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

   Vai a IAM

2. Attiva l'opzione Includi concessioni di ruoli fornite da Google.

3. Cerca l'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 seleziona il ruolo Creatore token account di servizio (roles/iam.serviceAccountTokenCreator).

7. Fai clic su Salva.

Crea un criterio in AWS

È necessario un criterio in AWS per consentire a Pub/Sub di eseguire l'autenticazione con AWS in modo che Pub/Sub possa importare i dati da un flusso di dati AWS Kinesis. Prima di creare un criterio AWS, crea uno stream di dati Kinesis e un consumer registrato. Ti consigliamo questa pratica in modo da poter limitare le autorizzazioni allo stream specifico.

• Per ulteriori informazioni su come creare uno stream di dati AWS Kinesis, consulta Stream di dati Kinesis.

• Per ulteriori informazioni sull'API AWS Kinesis stream di dati utilizzata per registrare i consumer, consulta RegisterStreamConsumer.

• Per ulteriori metodi e informazioni su come creare un criterio in AWS, consulta Creazione di criteri IAM.

Per creare un criterio 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 Access Management > Criteri.

3. Fai clic su Crea criterio.

4. In Seleziona un servizio, scegli Kinesis.

5. Per azione consentita, seleziona quanto segue:

   • Elenco > ListShards.

     Questa azione concede l'autorizzazione per elencare gli shard in un flusso e fornisce informazioni su ciascuno shard.

   • Read (Leggi) > SubscribeToShard.

     Questa azione concede l'autorizzazione per ascoltare uno shard specifico con fan-out migliorato.

   • Leggi > DescribeStreamConsumer.

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

   Queste autorizzazioni coprono la lettura dallo stream. Pub/Sub supporta solo la lettura da un flusso Kinesis con Enhanced Fan-Out, utilizzando l'API SubscribeToShard per i flussi di dati.

6. In Risorse, se vuoi limitare il criterio a uno specifico flusso o consumer (consigliato), specifica l'ARN del consumatore e l'ARN dello stream.

7. Fai clic su Aggiungi altre autorizzazioni.

8. In Seleziona un servizio, inserisci e seleziona STS.

9. Per azione consentita, seleziona Write > AssumeRoleWithWebIdentity.

   Questa azione concede l'autorizzazione per ottenere un set di credenziali di sicurezza temporanee per Pub/Sub per l'autenticazione nel flusso di dati di Kinesis utilizzando la federazione delle identità.

10. Tocca Avanti.

11. Inserisci un nome e una descrizione per il criterio.

12. Fai clic su Crea criterio.

Crea un ruolo in AWS utilizzando un criterio di attendibilità personalizzato

Devi creare un ruolo in AWS in modo che Pub/Sub possa eseguire l'autenticazione in AWS per importare i dati da Kinesis Data Streams.

Per creare un ruolo utilizzando un criterio di attendibilità personalizzato, 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 Ruoli.

3. Fai clic su Crea ruolo.

4. In Seleziona entità attendibile, scegli Criterio di attendibilità personalizzati.

5. Nella sezione Criterio di attendibilità personalizzato, 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 dell'account di servizio che hai registrato in Registrare l'ID univoco dell'account di servizio.

6. Tocca Avanti.

7. In Aggiungi autorizzazioni, cerca e seleziona il criterio personalizzato che hai appena creato.

8. Tocca Avanti.

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

10. Fai clic su Crea ruolo.

Aggiungere il ruolo di publisher Pub/Sub all'account di servizio Pub/Sub

Devi assegnare un ruolo di editore all'account di servizio Pub/Sub in modo che Pub/Sub possa pubblicare nell'argomento di importazione dai flussi di dati di AWS Kinesis.

• Per attivare la pubblicazione da tutti gli argomenti di un progetto, consulta Abilitare la pubblicazione da tutti gli argomenti.

• Per attivare la pubblicazione da un argomento specifico (opzione consigliata), vedi Abilitare la pubblicazione da un singolo argomento.

Attiva la pubblicazione da tutti gli argomenti

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

   Vai a IAM

2. Attiva l'opzione Includi concessioni di ruoli fornite da Google.

3. Cerca l'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 seleziona il ruolo di publisher Pub/Sub (roles/pubsub.publisher).

7. Fai clic su Salva.

Attiva la pubblicazione da un singolo argomento

Se vuoi concedere l'autorizzazione di pubblicazione solo a uno specifico argomento di importazione, segui questi passaggi:

1. Nella console Google Cloud, attiva Cloud Shell.

   Attiva Cloud Shell

   Nella parte inferiore della console Google Cloud viene avviata una sessione di Cloud Shell che mostra un prompt della riga di comando. Cloud Shell è un ambiente shell con Google Cloud CLI già installato e con valori già impostati per il progetto attuale. L'inizializzazione della sessione può richiedere alcuni secondi.

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 dell'argomento di importazione.

   • PROJECT_NUMBER è il numero del progetto. Per visualizzare il numero del progetto, consulta Identificazione dei progetti.

Aggiungere il ruolo utente dell'account di servizio all'account di servizio

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

Svolgi i seguenti passaggi:

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

   Vai a IAM

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

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

4. Cerca e seleziona il ruolo Utente account di servizio (roles/iam.serviceAccountUser).

5. Fai clic su Salva.

Creare un argomento di importazione

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

Assicurati di aver completato le seguenti procedure:

• Configura l'identità federata per accedere a Kinesis Data Streams

• Aggiungere il ruolo di publisher Pub/Sub all'account di servizio Pub/Sub

• Aggiungere il ruolo utente dell'account di servizio all'account di servizio

Per creare un argomento di importazione:

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 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.`);
}

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 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.`);
}

Per ulteriori informazioni sugli ARN, consulta la pagina relativa ai nomi delle risorse Amazon (ARN) e agli identificatori IAM.

Se riscontri problemi, consulta l'articolo Risolvere i problemi relativi all'importazione.

Modificare un argomento di importazione

Puoi modificare le impostazioni dell'origine dati di importazione di un argomento di importazione. Svolgi i seguenti passaggi:

Quote e limiti per gli argomenti di importazione

La velocità effettiva del publisher per gli argomenti di importazione è vincolata dalla quota di pubblicazione dell'argomento. Per maggiori informazioni, consulta Quote e limiti di Pub/Sub.

Passaggi successivi

• Monitorare un argomento relativo all'importazione.

• Scegli il tipo di sottoscrizione per l'argomento.

• Scopri come pubblicare un messaggio in un argomento.

• Crea o modifica un argomento con gcloud CLI, le API REST o le librerie client.

• Risolvere i problemi relativi all'importazione.

• Passaggi per configurare il job di trasferimento dello spazio di archiviazione al fine di trasferire i dati da AWS S3 a GCS.