Mengubah jenis topik

Anda dapat mengonversi topik impor menjadi topik standar atau sebaliknya, topik standar menjadi topik impor.

Mengonversi topik impor menjadi topik standar

Untuk mengonversi topik impor menjadi topik standar, hapus setelan penyerapan. Lakukan langkah-langkah berikut:

Konsol

  1. Di Google Cloud konsol, buka halaman Topics.

    Buka Topik

  2. Klik topik impor.

  3. Di halaman detail topik, klik Edit.

  4. Hapus centang pada opsi Enable ingestion.

  5. Klik Update.

gcloud

  1. Di konsol, aktifkan Cloud Shell. Google Cloud

    Aktifkan Cloud Shell

    Di bagian bawah konsol Google Cloud , sesi Cloud Shell akan dimulai dan menampilkan prompt command line. Cloud Shell adalah lingkungan shell dengan Google Cloud CLI yang sudah terinstal, dan dengan nilai yang sudah ditetapkan untuk project Anda saat ini. Diperlukan waktu beberapa detik untuk melakukan inisialisasi pada sesi.

  2. Jalankan perintah gcloud pubsub topics update:

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

    Ganti TOPIC_ID dengan ID topik.

Mengonversi topik standar menjadi topik impor Amazon Kinesis Data Streams

Untuk mengonversi topik standar menjadi topik impor Amazon Kinesis Data Streams, periksa terlebih dahulu apakah Anda memenuhi semua prasyarat.

Konsol

  1. Di Google Cloud konsol, buka halaman Topics.

    Buka Topik

  2. Klik topik yang ingin Anda konversi menjadi topik impor.

  3. Di halaman detail topik, klik Edit.

  4. Pilih opsi Enable ingestion.

  5. Untuk sumber penyerapan, pilih Amazon Kinesis Data Streams.

  6. Masukkan detail berikut:

    • Kinesis Stream ARN: ARN untuk Kinesis Data Stream yang akan Anda serap ke Pub/Sub. Format ARN adalah sebagai berikut: arn:${Partition}:kinesis:${Region}:${Account}:stream/${StreamName}.

    • Kinesis Consumer ARN: ARN resource konsumen yang terdaftar ke AWS Kinesis Data Stream. Format ARN adalah sebagai berikut: arn:${Partition}:kinesis:${Region}:${Account}:${StreamType}/${StreamName}/consumer/${ConsumerName}:${ConsumerCreationTimpstamp}.

    • AWS Role ARN: ARN peran AWS. Format ARN peran adalah sebagai berikut: arn:aws:iam::${Account}:role/${RoleName}.

    • Akun layanan: Akun layanan yang Anda buat.

  7. Klik Update.

gcloud

  1. Di konsol, aktifkan Cloud Shell. Google Cloud

    Aktifkan Cloud Shell

    Di bagian bawah konsol Google Cloud , sesi Cloud Shell akan dimulai dan menampilkan prompt command line. Cloud Shell adalah lingkungan shell dengan Google Cloud CLI yang sudah terinstal, dan dengan nilai yang sudah ditetapkan untuk project Anda saat ini. Diperlukan waktu beberapa detik untuk melakukan inisialisasi pada sesi.

  2. Jalankan perintah gcloud pubsub topics update dengan semua flag yang disebutkan dalam contoh berikut:

    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

    Ganti kode berikut:

    • TOPIC_ID adalah ID atau nama topik. Kolom ini tidak dapat diperbarui.

    • KINESIS_STREAM_ARN adalah ARN untuk Kinesis Data Streams yang akan Anda serap ke Pub/Sub. Format ARN adalah sebagai berikut: arn:${Partition}:kinesis:${Region}:${Account}:stream/${StreamName}.

    • KINESIS_CONSUMER_ARN adalah ARN resource konsumen yang terdaftar ke AWS Kinesis Data Streams. Format ARN adalah sebagai berikut: arn:${Partition}:kinesis:${Region}:${Account}:${StreamType}/${StreamName}/consumer/${ConsumerName}:${ConsumerCreationTimpstamp}.

    • KINESIS_ROLE_ARN adalah ARN peran AWS. Format ARN peran adalah sebagai berikut: arn:aws:iam::${Account}:role/${RoleName}.

    • PUBSUB_SERVICE_ACCOUNT adalah akun layanan yang Anda dibuat.

Go

Contoh berikut menggunakan versi utama library klien Pub/Sub Go (v2). Jika Anda masih menggunakan library v1, lihat panduan migrasi ke v2. Untuk melihat daftar contoh kode v1, lihat contoh kode yang tidak digunakan lagi.

Sebelum mencoba contoh ini, ikuti petunjuk penyiapan Go di Panduan memulai: Menggunakan Library Klien. Untuk mengetahui informasi selengkapnya, lihat dokumentasi referensi 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

Sebelum mencoba contoh ini, ikuti petunjuk penyiapan Java di Panduan memulai: Menggunakan Library Klien. Untuk mengetahui informasi selengkapnya, lihat dokumentasi referensi 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

Sebelum mencoba contoh ini, ikuti petunjuk penyiapan Node.js di Panduan memulai: Menggunakan Library Klien. Untuk mengetahui informasi selengkapnya, lihat dokumentasi referensi 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

Sebelum mencoba contoh ini, ikuti petunjuk penyiapan Python di Panduan memulai: Menggunakan Library Klien. Untuk mengetahui informasi selengkapnya, lihat dokumentasi referensi 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++

Sebelum mencoba contoh ini, ikuti petunjuk penyiapan C++ di Panduan memulai: Menggunakan Library Klien. Untuk mengetahui informasi selengkapnya, lihat dokumentasi referensi 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

Sebelum mencoba contoh ini, ikuti petunjuk penyiapan Node.js di Panduan memulai: Menggunakan Library Klien. Untuk mengetahui informasi selengkapnya, lihat dokumentasi referensi 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.');
}

Untuk mengetahui informasi selengkapnya tentang ARN, lihat Nama Resource Amazon (ARN) dan ID IAM.

Mengonversi topik standar menjadi topik impor Cloud Storage

Untuk mengonversi topik standar menjadi topik impor Cloud Storage, periksa terlebih dahulu apakah Anda memenuhi semua prasyarat.

Konsol

  1. Di Google Cloud konsol, buka halaman Topics.

    Buka Topik

  2. Klik topik yang ingin Anda konversi menjadi topik impor Cloud Storage.

  3. Di halaman detail topik, klik Edit.

  4. Pilih opsi Enable ingestion.

  5. Untuk sumber penyerapan, pilih Google Cloud Storage.

  6. Untuk bucket Cloud Storage, klik Browse.

    Halaman Select bucket akan terbuka. Pilih salah satu opsi berikut:

    • Pilih bucket yang ada dari project yang sesuai.

    • Klik ikon buat dan ikuti petunjuk di layar untuk membuat bucket baru. Setelah membuat bucket, pilih bucket untuk topik impor Cloud Storage.

  7. Saat Anda menentukan bucket, Pub/Sub akan memeriksa izin yang sesuai pada bucket untuk akun layanan Pub/Sub. Jika ada masalah izin, Anda akan melihat pesan error terkait izin.

    Jika Anda mengalami masalah izin, klik Set permissions. Untuk mengetahui informasi selengkapnya, lihat Memberikan izin Cloud Storage ke akun layanan Pub/Sub.

  8. Untuk Object format, pilih Text, Avro, atau Pub/Sub Avro.

    Jika memilih Text, Anda dapat secara opsional menentukan Delimiter dengan mana untuk membagi objek menjadi pesan.

    Untuk mengetahui informasi selengkapnya tentang opsi ini, lihat Format input.

  9. Opsional. Anda dapat menentukan Minimum object creation time untuk topik Anda. Jika ditetapkan, hanya objek yang dibuat setelah waktu pembuatan objek minimum yang akan diserap.

    Untuk mengetahui informasi selengkapnya, lihat Waktu pembuatan objek minimum.

  10. Anda harus menentukan Glob pattern. Untuk menyerap semua objek dalam bucket, gunakan ** sebagai pola glob. Hanya objek yang cocok dengan pola yang diberikan yang akan diserap.

    Untuk mengetahui informasi selengkapnya, lihat Mencocokkan pola glob.

  11. Pertahankan setelan default lainnya.
  12. Klik Update topic.

gcloud

  1. Di konsol, aktifkan Cloud Shell. Google Cloud

    Aktifkan Cloud Shell

    Di bagian bawah konsol Google Cloud , sesi Cloud Shell akan dimulai dan menampilkan prompt command line. Cloud Shell adalah lingkungan shell dengan Google Cloud CLI yang sudah terinstal, dan dengan nilai yang sudah ditetapkan untuk project Anda saat ini. Diperlukan waktu beberapa detik untuk melakukan inisialisasi pada sesi.

  2. Agar tidak kehilangan setelan untuk topik impor, pastikan untuk menyertakan semuanya setiap kali Anda memperbarui topik. Jika Anda tidak menyertakan sesuatu, Pub/Sub akan mereset setelan ke nilai default aslinya.

    Jalankan perintah gcloud pubsub topics update dengan semua flag yang disebutkan dalam contoh berikut:

    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

    Ganti kode berikut:

    • TOPIC_ID adalah ID atau nama topik. Kolom ini tidak dapat diperbarui.

    • BUCKET_NAME: Menentukan nama bucket yang ada. Contohnya, prod_bucket. Nama bucket tidak boleh menyertakan ID project. Untuk membuat bucket, lihat Membuat bucket.

    • INPUT_FORMAT: Menentukan format objek yang diserap. Ini dapat berupa text, avro, atau pubsub_avro. Untuk mengetahui informasi selengkapnya tentang opsi ini, lihat Format input.

    • TEXT_DELIMITER: Menentukan pemisah untuk membagi objek teks menjadi pesan Pub/Sub. Pemisah ini harus berupa satu karakter dan hanya boleh ditetapkan jika INPUT_FORMAT adalah text. Nilai defaultnya adalah karakter baris baru (\n).

      Saat menggunakan gcloud CLI untuk menentukan pemisah, perhatikan penanganan karakter khusus seperti baris baru \n. Gunakan format '\n' untuk memastikan pemisah ditafsirkan dengan benar. Hanya menggunakan \n tanpa tanda kutip atau karakter escape akan menghasilkan pemisah "n".

    • MINIMUM_OBJECT_CREATE_TIME: Menentukan waktu minimum saat objek dibuat agar dapat diserap. Waktu ini harus dalam UTC dengan format YYYY-MM-DDThh:mm:ssZ. Contohnya, 2024-10-14T08:30:30Z.

      Tanggal apa pun, masa lalu atau masa mendatang, dari 0001-01-01T00:00:00Z hingga 9999-12-31T23:59:59Z inklusif, valid.

    • MATCH_GLOB: Menentukan pola glob yang akan dicocokkan agar objek dapat diserap. Saat Anda menggunakan gcloud CLI, glob yang cocok dengan karakter * harus memiliki karakter * yang diformat sebagai karakter escape dalam bentuk \*\*.txt atau seluruh glob yang cocok harus dalam tanda kutip "**.txt" atau '**.txt'. Untuk mengetahui informasi tentang sintaksis yang didukung untuk pola glob, lihat dokumentasi Cloud Storage.