Membuat topik impor Amazon Kinesis Data Streams

Topik impor Amazon Kinesis Data Streams memungkinkan Anda terus-menerus menyerap data dari Amazon Kinesis Data Streams sebagai sumber eksternal dan ke Pub/Sub. Kemudian, Anda dapat mengalirkan data ke tujuan mana pun yang didukung Pub/Sub.

Untuk mengetahui informasi selengkapnya tentang topik impor, lihat Tentang topik impor.

Sebelum memulai

• Ketahui lebih lanjut proses penerbitan Pub/Sub.

• Konfigurasi peran dan izin yang diperlukan untuk mengelola topik impor Amazon Kinesis Data Streams, termasuk yang berikut:

  • Tambahkan peran Pub/Sub Publisher ke akun layanan Pub/Sub

  • Tambahkan peran agen layanan Pub/Sub ke akun layanan Pub/Sub, jika belum diberikan.

  • Menambahkan peran pengguna akun layanan ke akun layanan

• Siapkan workload identity federation agar Trusted Cloud dapat mengakses layanan streaming eksternal.

Peran dan izin yang diperlukan

Untuk mendapatkan izin yang Anda perlukan guna membuat dan mengelola topik impor Amazon Kinesis Data Streams, minta administrator untuk memberi Anda peran IAM Pub/Sub Editor (roles/pubsub.editor) di topik atau project Anda. Untuk mengetahui informasi selengkapnya tentang cara memberikan peran, lihat Mengelola akses ke project, folder, dan organisasi.

Peran bawaan ini berisi izin yang diperlukan untuk membuat dan mengelola topik impor Amazon Kinesis Data Streams. Untuk melihat izin yang benar-benar diperlukan, luaskan bagian Izin yang diperlukan:

Izin yang diperlukan

Izin berikut diperlukan untuk membuat dan mengelola topik impor Amazon Kinesis Data Streams:

• Buat topik impor: pubsub.topics.create
• Menghapus topik impor: pubsub.topics.delete
• Dapatkan topik impor: pubsub.topics.get
• Cantumkan topik impor: pubsub.topics.list
• Memublikasikan ke topik impor: pubsub.topics.publish and pubsub.serviceAgent
• Perbarui topik impor: pubsub.topics.update
• Dapatkan kebijakan IAM untuk topik impor: pubsub.topics.getIamPolicy
• Konfigurasi kebijakan IAM untuk topik impor: pubsub.topics.setIamPolicy

Anda mungkin juga bisa mendapatkan izin ini dengan peran khusus atau peran bawaan lainnya.

Anda dapat mengonfigurasi kontrol akses di level project dan level resource individual.

Menyiapkan identitas gabungan untuk mengakses Amazon Kinesis Data Streams

Workload Identity Federation memungkinkan Trusted Cloud layanan mengakses beban kerja yang berjalan di luar Trusted Cloud. Dengan federasi identitas, Anda tidak perlu memelihara atau meneruskan kredensial ke Trusted Cloud untuk mengakses resource Anda di cloud lain. Sebagai gantinya, Anda dapat menggunakan identitas workload itu sendiri untuk melakukan autentikasi ke Trusted Cloud dan mengakses resource.

Buat akun layanan di Trusted Cloud

Ini langkah opsional. Jika sudah memiliki akun layanan, Anda dapat menggunakannya dalam prosedur ini, bukan membuat akun layanan baru. Jika Anda menggunakan akun layanan yang sudah ada, buka Mencatat ID unik akun layanan untuk langkah berikutnya.

Untuk topik impor Amazon Kinesis Data Streams, Pub/Sub menggunakan akun layanan sebagai identitas untuk mengakses resource dari AWS.

Untuk mengetahui informasi selengkapnya tentang cara membuat akun layanan, termasuk prasyarat, peran dan izin yang diperlukan, serta panduan penamaan, lihat Membuat akun layanan. Setelah membuat akun layanan, Anda mungkin perlu menunggu selama 60 detik atau lebih sebelum menggunakan akun layanan tersebut. Perilaku ini terjadi karena operasi baca pada akhirnya konsisten; perlu waktu hingga akun layanan baru terlihat.

Mencatat ID unik akun layanan

Anda memerlukan ID unik akun layanan untuk menyiapkan peran di AWS.

1. Di konsol Trusted Cloud , buka halaman detail Service account.

   Buka akun layanan

2. Klik akun layanan yang baru saja Anda buat atau yang akan Anda gunakan.

3. Dari halaman Service account details, catat nomor ID Unik.

   Anda memerlukan ID sebagai bagian dari alur kerja untuk menyiapkan peran di AWS.

Tambahkan peran pembuat token akun layanan ke akun layanan Pub/Sub

Peran pembuat token akun layanan (roles/iam.serviceAccountTokenCreator) memungkinkan akun utama membuat kredensial berumur pendek untuk akun layanan. Token atau kredensial ini digunakan untuk meniru identitas akun layanan.

Untuk informasi selengkapnya tentang peniruan akun layanan, lihat Peniruan akun layanan.

Anda juga dapat menambahkan peran penerbit Pub/Sub (roles/pubsub.publisher) selama prosedur ini. Untuk mengetahui informasi selengkapnya tentang peran dan alasan Anda menambahkannya, lihat Menambahkan peran Pub/Sub Publisher ke akun layanan Pub/Sub.

1. Di konsol Trusted Cloud , buka halaman IAM.

   Buka IAM

2. Centang kotak Include S3NS-provided role grants.

3. Cari akun layanan yang memiliki format service-{PROJECT_NUMBER}@gcp-sa-pubsub.s3ns-system.iam.gserviceaccount.com.

4. Untuk akun layanan ini, klik tombol Edit Principal.

5. Jika diperlukan, klik Tambahkan peran lain.

6. Cari dan klik Service account token creator role (roles/iam.serviceAccountTokenCreator).

7. Klik Simpan.

Membuat kebijakan di AWS

Anda memerlukan kebijakan di AWS agar Pub/Sub dapat melakukan autentikasi ke AWS sehingga Pub/Sub dapat menyerap data dari Amazon Kinesis Data Streams.

• Untuk mengetahui metode dan informasi selengkapnya tentang cara membuat kebijakan di AWS, lihat Membuat kebijakan IAM.

Untuk membuat kebijakan di AWS, lakukan langkah-langkah berikut:

1. Login ke AWS Management Console, lalu buka konsol IAM.

2. Di panel navigasi konsol untuk IAM, klik Access Management > Policies.

3. Klik Create policy.

4. Untuk Pilih layanan, klik Kinesis.

5. Untuk Tindakan yang diizinkan, klik berikut ini:

   • List > ListShards.

     Tindakan ini memberikan izin untuk mencantumkan pecahan dalam aliran dan memberikan informasi tentang setiap pecahan.

   • Read > SubscribeToShard.

     Tindakan ini memberikan izin untuk memproses shard tertentu dengan fan-out yang ditingkatkan.

   • Baca > DescribeStreamConsumer.

     Tindakan ini memberikan izin untuk mendapatkan deskripsi konsumen stream terdaftar.

   Izin ini mencakup pembacaan dari aliran. Pub/Sub hanya mendukung pembacaan dari stream Kinesis dengan Enhanced Fan-Out menggunakan streaming SubscribeToShard API.

6. Untuk Resources, jika Anda ingin membatasi kebijakan ke aliran atau konsumen tertentu (direkomendasikan), tentukan ARN konsumen dan ARN aliran.

7. Klik Tambahkan izin lainnya.

8. Untuk Select a service, klik STS.

9. Untuk Action allowed, klik Write > AssumeRoleWithWebIdentity.

   Tindakan ini memberikan izin untuk mendapatkan serangkaian kredensial keamanan sementara untuk Pub/Sub agar dapat melakukan autentikasi ke Amazon Kinesis Data Streams dengan menggunakan federasi identitas.

10. Klik Berikutnya.

11. Masukkan nama dan deskripsi kebijakan.

12. Klik Create policy.

Membuat peran di AWS menggunakan kebijakan kepercayaan kustom

Anda harus membuat peran di AWS agar Pub/Sub dapat melakukan autentikasi ke AWS untuk menyerap data dari Amazon Kinesis Data Streams.

1. Login ke AWS Management Console, lalu buka konsol IAM.

2. Di panel navigasi konsol untuk IAM, klik Roles.

3. Klik Buat peran.

4. Untuk Pilih entitas tepercaya, klik Kebijakan kepercayaan kustom.

5. Di bagian Custom trust policy, masukkan atau tempelkan berikut:

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

   Ganti <SERVICE_ACCOUNT_UNIQUE_ID> dengan ID unik akun layanan yang Anda catat di Mencatat ID unik akun layanan.

6. Klik Berikutnya.

7. Untuk Tambahkan izin, telusuri dan klik kebijakan kustom yang baru saja Anda buat.

8. Klik Berikutnya.

9. Masukkan nama dan deskripsi peran.

10. Klik Buat peran.

Tambahkan peran penerbit Pub/Sub ke akun utama Pub/Sub

Untuk mengaktifkan publikasi, Anda harus menetapkan peran publisher ke akun layanan Pub/Sub agar Pub/Sub dapat memublikasikan ke topik impor Amazon Kinesis Data Streams.

Menambahkan peran agen layanan Pub/Sub ke akun layanan Pub/Sub

Agar Pub/Sub dapat menggunakan kuota publikasi project topik impor Anda, agen layanan Pub/Sub memerlukan izin serviceusage.services.use di project topik impor Anda.

Untuk memberikan izin ini, sebaiknya tambahkan peran agen layanan Pub/Sub ke akun layanan Pub/Sub.

Jika akun layanan Pub/Sub tidak memiliki peran agen layanan Pub/Sub, peran tersebut dapat diberikan sebagai berikut:

1. Di konsol Trusted Cloud , buka halaman IAM.

   Buka IAM

2. Centang kotak Include S3NS-provided role grants.

3. Cari akun layanan yang memiliki format service-{PROJECT_NUMBER}@gcp-sa-pubsub.s3ns-system.iam.gserviceaccount.com.

4. Untuk akun layanan ini, klik tombol Edit principal.

5. Jika diperlukan, klik Tambahkan peran lain.

6. Cari dan klik peran Pub/Sub Service Agent (roles/pubsub.serviceAgent).

7. Klik Simpan.

Mengaktifkan publikasi dari semua topik

Gunakan metode ini jika Anda belum membuat topik impor Amazon Kinesis Data Streams.

1. Di konsol Trusted Cloud , buka halaman IAM.

   Buka IAM

2. Centang kotak Include S3NS-provided role grants.

3. Cari akun layanan yang memiliki format service-{PROJECT_NUMBER}@gcp-sa-pubsub.s3ns-system.iam.gserviceaccount.com.

4. Untuk akun layanan ini, klik tombol Edit principal.

5. Jika diperlukan, klik Tambahkan peran lain.

6. Cari dan klik Pub/Sub publisher role (roles/pubsub.publisher).

7. Klik Simpan.

Mengaktifkan publikasi dari satu topik

Gunakan metode ini hanya jika topik impor Amazon Kinesis Data Streams sudah ada.

1. In the Trusted Cloud console, activate Cloud Shell.

   Activate Cloud Shell

   At the bottom of the Trusted 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. Jalankan perintah gcloud pubsub topics add-iam-policy-binding:

   gcloud pubsub topics add-iam-policy-binding TOPIC_ID \
      --member="serviceAccount:service-PROJECT_NUMBER@gcp-sa-pubsub.s3ns-system.iam.gserviceaccount.com" \
      --role="roles/pubsub.publisher"

   Ganti kode berikut:

   • TOPIC_ID: ID topik dari topik impor Amazon Kinesis Data Streams.

   • PROJECT_NUMBER: nomor project. Untuk melihat nomor project, lihat Mengidentifikasi project.

Menambahkan peran pengguna akun layanan ke akun layanan

Peran Pengguna Akun Layanan (roles/iam.serviceAccountUser) mencakup izin iam.serviceAccounts.actAs yang memungkinkan akun utama melampirkan akun layanan ke setelan penyerapan topik impor Amazon Kinesis Data Streams dan menggunakan akun layanan tersebut untuk identitas gabungan.

1. Di konsol Trusted Cloud , buka halaman IAM.

   Buka IAM

2. Untuk akun utama yang mengeluarkan panggilan buat atau perbarui topik, klik tombol Edit akun utama.

3. Jika diperlukan, klik Tambahkan peran lain.

4. Cari dan klik Service account user role (roles/iam.serviceAccountUser).

5. Klik Simpan.

Menggunakan topik Amazon Kinesis Data Streams

Anda dapat membuat topik impor baru atau mengedit topik yang ada.

Pertimbangan

• Membuat topik dan langganan secara terpisah, meskipun dilakukan secara berurutan dengan cepat, dapat menyebabkan hilangnya data. Ada periode singkat saat topik ada tanpa langganan. Jika ada data yang dikirim ke topik selama waktu ini, data tersebut akan hilang. Dengan membuat topik terlebih dahulu, membuat langganan, lalu mengonversi topik menjadi topik impor, Anda menjamin tidak ada pesan yang terlewat selama proses impor.

Membuat topik impor Amazon Kinesis Data Streams

Untuk mengetahui lebih lanjut properti yang terkait dengan topik, lihat Properti topik.

Pastikan Anda telah menyelesaikan prosedur berikut:

• Menyiapkan identitas gabungan untuk mengakses Kinesis Data Streams

• Tambahkan peran Pub/Sub Publisher ke akun layanan Pub/Sub

• Menambahkan peran pengguna akun layanan ke akun layanan

Untuk membuat topik impor Amazon Kinesis Data Streams, ikuti langkah-langkah berikut:

Konsol

1. Di konsol Trusted Cloud , buka halaman Topics.

   Buka Topik

2. Klik Create topic.

3. Di kolom Topic ID, masukkan ID untuk topik impor Amazon Kinesis Data Streams Anda.

   Untuk mengetahui informasi selengkapnya tentang penamaan topik, lihat panduan penamaan.

4. Pilih Tambahkan langganan default.

5. Pilih Aktifkan penyerapan.

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

7. Masukkan detail berikut:

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

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

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

   • Akun layanan: Akun layanan yang Anda buat di Membuat akun layanan di Trusted Cloud.

8. Klik Create topic.

gcloud

1. In the Trusted Cloud console, activate Cloud Shell.

   Activate Cloud Shell

   At the bottom of the Trusted 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. Jalankan perintah 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
   

   Ganti kode berikut:

   • TOPIC_ID: ID topik.
   • KINESIS_STREAM_ARN: 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: ARN resource konsumen yang terdaftar ke AWS Kinesis Data Streams. Format ARN adalah sebagai berikut: arn:${Partition}:kinesis:${Region}:${Account}:${StreamType}/${StreamName}/consumer/${ConsumerName}:${ConsumerCreationTimestamp}.
   • KINESIS_ROLE_ARN: ARN peran AWS. Format ARN peran adalah sebagai berikut: arn:aws:iam::${Account}:role/${RoleName}.
   • PUBSUB_SERVICE_ACCOUNT: Akun layanan yang Anda buat di Membuat akun layanan di Google Cloud.

C++

Sebelum mencoba contoh ini, ikuti petunjuk penyiapan C++ di Panduan memulai: Menggunakan Library Klien. Untuk mengetahui informasi selengkapnya, lihat dokumentasi referensi Pub/Sub C++ API.

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

Contoh berikut menggunakan library klien Go Pub/Sub versi utama (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 Pub/Sub Go API.

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

Sebelum mencoba contoh ini, ikuti petunjuk penyiapan Java di Panduan memulai: Menggunakan Library Klien. Untuk mengetahui informasi selengkapnya, lihat dokumentasi referensi Pub/Sub Java API.

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

Sebelum mencoba contoh ini, ikuti petunjuk penyiapan Node.js di Panduan memulai: Menggunakan Library Klien. Untuk mengetahui informasi selengkapnya, lihat dokumentasi referensi API Node.js Pub/Sub.

/**
 * 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

Sebelum mencoba contoh ini, ikuti petunjuk penyiapan Node.js di Panduan memulai: Menggunakan Library Klien. Untuk mengetahui informasi selengkapnya, lihat dokumentasi referensi API Node.js Pub/Sub.

/**
 * 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

Sebelum mencoba contoh ini, ikuti petunjuk penyiapan Python di Panduan memulai: Menggunakan Library Klien. Untuk mengetahui informasi selengkapnya, lihat dokumentasi referensi Pub/Sub Python API.

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

Untuk mengetahui informasi selengkapnya tentang ARN, lihat Amazon Resource Names (ARN) dan IAM Identifiers.

Jika Anda mengalami masalah, lihat Memecahkan masalah topik impor Amazon Kinesis Data Streams.

Mengedit topik impor Amazon Kinesis Data Streams

Anda dapat mengedit setelan sumber data penyerapan topik impor Amazon Kinesis Data Streams. Lakukan langkah-langkah berikut:

Konsol

1. Di konsol Trusted Cloud , buka halaman Topics.

   Buka Topik

2. Klik topik impor Amazon Kinesis Data Streams.

3. Di halaman detail topik, klik Edit.

4. Perbarui kolom yang ingin Anda ubah.

5. Klik Perbarui.

gcloud

1. In the Trusted Cloud console, activate Cloud Shell.

   Activate Cloud Shell

   At the bottom of the Trusted 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. 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 
   
       --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 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}:${ConsumerCreationTimestamp}.

   • 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 buat di Membuat akun layanan di Trusted Cloud.

Kuota dan batas untuk topik impor Amazon Kinesis Data Streams

Throughput penayang untuk topik impor dibatasi oleh kuota publikasi topik. Untuk mengetahui informasi selengkapnya, lihat Kuota dan batas Pub/Sub.

Langkah berikutnya

• Memantau topik impor.

• Pilih jenis langganan untuk topik Anda.

• Pelajari cara memublikasikan pesan ke topik.

• Buat atau ubah topik dengan gcloud CLI, REST API, atau Library klien.

• Memecahkan masalah topik impor Amazon Kinesis Data Streams.

• Langkah-langkah untuk mengonfigurasi tugas transfer penyimpanan guna mentransfer data dari AWS S3 ke Cloud Storage.

Apache Kafka® adalah merek dagang terdaftar dari The Apache Software Foundation atau afiliasinya di Amerika Serikat dan/atau negara lain.