Modificare il tipo di argomento

Puoi convertire un argomento di importazione in uno standard o, viceversa, un argomento standard in uno di importazione.

Convertire un argomento di importazione in un argomento standard

Per convertire un argomento di importazione in un argomento standard, cancella le impostazioni di importazione. Segui questi passaggi:

Console

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

    Vai ad Argomenti

  2. Fai clic sull'argomento di importazione.

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

  4. Deseleziona l'opzione Attiva importazione.

  5. Fai clic su Aggiorna.

gcloud

  1. Nella Google Cloud console, attiva Cloud Shell.

    Attiva Cloud Shell

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

  2. Esegui il gcloud pubsub topics update comando:

    gcloud pubsub topics update TOPIC_ID \
    --clear-ingestion-data-source-settings

    Sostituisci TOPIC_ID con l'ID argomento.

Convertire un argomento standard in un argomento di importazione di Amazon Kinesis Data Streams

Per convertire un argomento standard in un argomento di importazione di Amazon Kinesis Data Streams, controlla innanzitutto di soddisfare tutti i prerequisiti.

Console

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

    Vai ad Argomenti

  2. Fai clic sull'argomento che vuoi convertire in un argomento di importazione.

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

  4. Seleziona l'opzione Attiva importazione.

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

  6. Inserisci i seguenti dettagli:

    • ARN del flusso Kinesis: l'ARN del flusso Kinesis Data Streams 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 che è 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.

  7. Fai clic su Aggiorna.

gcloud

  1. Nella Google Cloud console, attiva Cloud Shell.

    Attiva Cloud Shell

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

  2. Esegui il comando gcloud pubsub topics update con tutti i flag indicati nell'esempio seguente:

    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 o il nome dell'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 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.

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

	"cloud.google.com/go/pubsub/v2"
	"cloud.google.com/go/pubsub/v2/apiv1/pubsubpb"
	"google.golang.org/protobuf/types/known/fieldmaskpb"
)

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

	pbTopic := &pubsubpb.Topic{
		Name: topic,
		IngestionDataSourceSettings: &pubsubpb.IngestionDataSourceSettings{
			Source: &pubsubpb.IngestionDataSourceSettings_AwsKinesis_{
				AwsKinesis: &pubsubpb.IngestionDataSourceSettings_AwsKinesis{
					StreamArn:         streamARN,
					ConsumerArn:       consumerARN,
					AwsRoleArn:        awsRoleARN,
					GcpServiceAccount: gcpServiceAccount,
				},
			},
		},
	}
	updateReq := &pubsubpb.UpdateTopicRequest{
		Topic: pbTopic,
		UpdateMask: &fieldmaskpb.FieldMask{
			Paths: []string{"ingestion_data_source_settings"},
		},
	}
	topicCfg, err := client.TopicAdminClient.UpdateTopic(ctx, updateReq)
	if err != nil {
		return fmt.Errorf("topic.Update: %w", err)
	}
	fmt.Fprintf(w, "Topic updated with kinesis source: %v\n", topicCfg)
	return nil
}

Java

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


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

public class UpdateTopicTypeExample {
  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";

    UpdateTopicTypeExample.updateTopicTypeExample(
        projectId, topicId, streamArn, consumerArn, awsRoleArn, gcpServiceAccount);
  }

  public static void updateTopicTypeExample(
      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();

      // Construct the topic with Kinesis ingestion settings.
      Topic topic =
          Topic.newBuilder()
              .setName(topicName.toString())
              .setIngestionDataSourceSettings(ingestionDataSourceSettings)
              .build();

      // Construct a field mask to indicate which field to update in the topic.
      FieldMask updateMask =
          FieldMask.newBuilder().addPaths("ingestion_data_source_settings").build();

      UpdateTopicRequest request =
          UpdateTopicRequest.newBuilder().setTopic(topic).setUpdateMask(updateMask).build();

      Topic response = topicAdminClient.updateTopic(request);

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

Node.js

Prima di provare questo esempio, segui le istruzioni di configurazione di Node.js in 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 awsRoleArn = '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 updateTopicIngestionType(
  topicNameOrId,
  awsRoleArn,
  gcpServiceAccount,
  streamArn,
  consumerArn,
) {
  const metadata = {
    ingestionDataSourceSettings: {
      awsKinesis: {
        awsRoleArn,
        gcpServiceAccount,
        streamArn,
        consumerArn,
      },
    },
  };

  await pubSubClient.topic(topicNameOrId).setMetadata(metadata);

  console.log('Topic updated with Kinesis source successfully.');
}

Python

Prima di provare questo esempio, segui le istruzioni di configurazione di Python in 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
from google.pubsub_v1.types import UpdateTopicRequest
from google.protobuf import field_mask_pb2

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

update_request = UpdateTopicRequest(
    topic=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,
            )
        ),
    ),
    update_mask=field_mask_pb2.FieldMask(paths=["ingestion_data_source_settings"]),
)

topic = publisher.update_topic(request=update_request)
print(f"Updated topic: {topic.name} with AWS Kinesis Ingestion Settings")

C++

Prima di provare questo esempio, segui le istruzioni di configurazione di C++ in Guida rapida all'utilizzo delle librerie client. Per saperne di più, consulta la documentazione di riferimento dell'API 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::UpdateTopicRequest request;

  request.mutable_topic()->set_name(
      pubsub::Topic(std::move(project_id), std::move(topic_id)).FullName());
  auto* aws_kinesis = request.mutable_topic()
                          ->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);
  *request.mutable_update_mask()->add_paths() =
      "ingestion_data_source_settings";

  auto topic = client.UpdateTopic(request);
  if (!topic) throw std::move(topic).status();

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

Node.js

Prima di provare questo esempio, segui le istruzioni di configurazione di Node.js in 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 awsRoleArn = '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, TopicMetadata} from '@google-cloud/pubsub';

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

async function updateTopicIngestionType(
  topicNameOrId: string,
  awsRoleArn: string,
  gcpServiceAccount: string,
  streamArn: string,
  consumerArn: string,
) {
  const metadata: TopicMetadata = {
    ingestionDataSourceSettings: {
      awsKinesis: {
        awsRoleArn,
        gcpServiceAccount,
        streamArn,
        consumerArn,
      },
    },
  };

  await pubSubClient.topic(topicNameOrId).setMetadata(metadata);

  console.log('Topic updated with Kinesis source successfully.');
}

Per saperne di più sugli ARN, consulta Nomi risorsa Amazon (ARN) e identificatori IAM.

Convertire un argomento standard in un argomento di importazione di Cloud Storage

Per convertire un argomento standard in un argomento di importazione di Cloud Storage, controlla innanzitutto di soddisfare tutti i prerequisiti.

Console

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

    Vai ad Argomenti

  2. Fai clic sull'argomento che vuoi convertire in un argomento di importazione di Cloud Storage.

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

  4. Seleziona l'opzione Attiva importazione.

  5. Per l'origine di importazione, seleziona Google Cloud Storage.

  6. Per il bucket Cloud Storage, fai clic su Sfoglia.

    Si apre la pagina Seleziona bucket. Seleziona una delle seguenti opzioni:

    • Seleziona un bucket esistente da qualsiasi progetto appropriato.

    • Fai clic sull'icona di creazione e segui le istruzioni sullo schermo per creare un nuovo bucket. Dopo aver creato il bucket, selezionalo per l'argomento di importazione di Cloud Storage.

  7. Quando specifichi il bucket, Pub/Sub verifica le autorizzazioni appropriate sul bucket per il Pub/Sub account di servizio. Se si verificano problemi di autorizzazioni, viene visualizzato un messaggio di errore relativo alle autorizzazioni.

    Se si verificano problemi di autorizzazione, fai clic su Imposta autorizzazioni. Per maggiori informazioni, consulta Concedi autorizzazioni Cloud Storage al service account Pub/Sub.

  8. Per Formato oggetto, seleziona Testo, Avro o Pub/Sub Avro.

    Se selezioni Testo, puoi facoltativamente specificare un Delimitatore con cui dividere gli oggetti in messaggi.

    Per ulteriori informazioni su queste opzioni, vedi Formato di input.

  9. Facoltativo. Puoi specificare un Tempo minimo di creazione dell'oggetto per il tuo argomento. Se impostato, vengono inseriti solo gli oggetti creati dopo il tempo minimo di creazione dell'oggetto vengono inseriti.

    Per maggiori informazioni, consulta Tempo minimo di creazione dell'oggetto.

  10. Devi specificare un Pattern glob. Per importare tutti gli oggetti nel bucket, utilizza ** come pattern glob. Vengono importati solo gli oggetti che corrispondono al pattern specificato.

    Per saperne di più, vedi Crea una corrispondenza con un pattern glob.

  11. Mantieni le altre impostazioni predefinite.
  12. Fai clic su Aggiorna argomento.

gcloud

  1. Nella Google Cloud console, attiva Cloud Shell.

    Attiva Cloud Shell

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

  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 indicati nell'esempio seguente campione:

    gcloud pubsub topics update TOPIC_ID \
    --cloud-storage-ingestion-bucket=BUCKET_NAME\
    --cloud-storage-ingestion-input-format=INPUT_FORMAT\
    --cloud-storage-ingestion-text-delimiter=TEXT_DELIMITER\
    --cloud-storage-ingestion-minimum-object-create-time=MINIMUM_OBJECT_CREATE_TIME\
    --cloud-storage-ingestion-match-glob=MATCH_GLOB

    Sostituisci quanto segue:

    • TOPIC_ID è l'ID o il nome dell'argomento. Questo campo non può essere aggiornato.

    • BUCKET_NAME: specifica il nome di un bucket esistente. Ad esempio, prod_bucket. Il nome del bucket non deve includere l'ID progetto. Per creare un bucket, consulta Crea bucket.

    • INPUT_FORMAT: specifica il formato degli oggetti inseriti. Può essere text, avro o pubsub_avro. Per ulteriori informazioni su queste opzioni, vedi Formato di input.

    • TEXT_DELIMITER: specifica il delimitatore con cui dividere gli oggetti di testo in messaggi Pub/Sub. Deve essere un singolo carattere e deve essere impostato solo quando INPUT_FORMAT è text. Il valore predefinito è il carattere di nuova riga (\n).

      Quando utilizzi gcloud CLI per specificare il delimitatore, presta attenzione alla gestione dei caratteri speciali come la nuova riga \n. Utilizza il formato '\n' per assicurarti che il delimitatore venga interpretato correttamente. Se utilizzi semplicemente \n senza virgolette o sequenze di escape, il delimitatore sarà "n".

    • MINIMUM_OBJECT_CREATE_TIME: specifica il tempo minimo di creazione di un oggetto affinché venga importato. Deve essere in formato UTC YYYY-MM-DDThh:mm:ssZ. Ad esempio, 2024-10-14T08:30:30Z.

      È valida qualsiasi data, passata o futura, compresa tra 0001-01-01T00:00:00Z e 9999-12-31T23:59:59Z.

    • MATCH_GLOB: specifica il pattern glob con cui creare una corrispondenza affinché un oggetto venga importato. Quando utilizzi gcloud CLI, un glob di corrispondenza con caratteri * deve avere il carattere * formattato come sequenza di escape nella forma \*\*.txt oppure l'intero glob di corrispondenza deve essere racchiuso tra virgolette "**.txt" o '**.txt'. Per informazioni sulla sintassi supportata per i pattern glob, consulta la documentazione di Cloud Storage.