Amazon Kinesis Data Streams-Importthema erstellen

Mit einem Amazon Kinesis Data Streams-Importthema können Sie Daten aus Amazon Kinesis Data Streams als externe Quelle in Pub/Sub aufnehmen. Dann können Sie die Daten in jedes der Ziele streamen, Pub/Sub unterstützt.

In diesem Dokument erfahren Sie, wie Sie Importthemen für Amazon Kinesis Data Streams erstellen und verwalten. Informationen zum Erstellen eines Standardthemas finden Sie unter Standardthema erstellen.

Hinweis

• Weitere Informationen zum Pub/Sub-Veröffentlichungsprozess

• Konfigurieren Sie die erforderlichen Rollen und Berechtigungen zum Verwalten von Amazon Kinesis Data Streams-Importthemen, einschließlich der folgenden:

  • Dem Pub/Sub-Dienstkonto die Rolle „Pub/Sub-Publisher“ hinzufügen

  • Dem Dienstkonto die Rolle „Dienstkontonutzer“ hinzufügen

• Richten Sie die Workload Identity-Föderation ein, damit Google Cloud auf den externen Streamingdienst zugreifen kann.

Erforderliche Rollen und Berechtigungen zum Verwalten von Importthemen

Bitten Sie Ihren Administrator, Ihnen die IAM-Rolle „Pub/Sub-Bearbeiter“ (roles/pubsub.editor) für Ihr Thema oder Projekt zu gewähren, um die Berechtigungen zum Erstellen und Verwalten von Amazon Kinesis Data Streams-Importthemen zu erhalten. Weitere Informationen zum Zuweisen von Rollen finden Sie unter Zugriff auf Projekte, Ordner und Organisationen verwalten.

Diese vordefinierte Rolle enthält die Berechtigungen, die zum Erstellen und Verwalten von Amazon Kinesis Data Streams-Importthemen erforderlich sind. Erweitern Sie den Abschnitt Erforderliche Berechtigungen, um die erforderlichen Berechtigungen anzuzeigen:

Erforderliche Berechtigungen

Die folgenden Berechtigungen sind zum Erstellen und Verwalten von Amazon Kinesis Data Streams-Importthemen erforderlich:

• Erstellen Sie ein Importthema: pubsub.topics.create
• So löschen Sie ein Importthema: pubsub.topics.delete
• Importthema abrufen: pubsub.topics.get
• Importthema angeben: pubsub.topics.list
• In einem importierten Thema veröffentlichen: pubsub.topics.publish
• Aktualisieren Sie ein Importthema: pubsub.topics.update
• Rufen Sie die IAM-Richtlinie für ein Importthema ab: pubsub.topics.getIamPolicy
• Konfigurieren Sie die IAM-Richtlinie für ein Importthema: pubsub.topics.setIamPolicy

Sie können diese Berechtigungen auch mit benutzerdefinierten Rollen oder anderen vordefinierten Rollen erhalten.

Sie können die Zugriffssteuerung auf Projektebene und auf der auf der Ebene einzelner Ressourcen.

Föderierte Identität für den Zugriff auf Kinesis Data Streams einrichten

Mit der Workload Identity-Föderation können Google Cloud-Dienste auf Arbeitslasten zugreifen, die außerhalb von Google Cloud ausgeführt werden. Mit der Identitätsfederation müssen Sie keine Anmeldedaten für Google Cloud verwalten oder an Google Cloud weitergeben, um auf Ihre Ressourcen in anderen Clouds zuzugreifen. Stattdessen können Sie die Identitäten der Arbeitslasten selbst um sich bei Google Cloud zu authentifizieren und auf Ressourcen zuzugreifen.

Dienstkonto in Google Cloud erstellen

Dieser Schritt ist optional. Wenn Sie bereits ein Dienstkonto haben, können Sie es in diesem Verfahren verwenden, anstatt ein neues Dienstkonto zu erstellen. Wenn Sie ein vorhandenes Dienstkonto verwenden, gehen Sie zu Notieren Sie sich die eindeutige Dienstkonto-ID für die nächsten Schritt.

Bei einem Amazon Kinesis Data Streams-Importthema verwendet Pub/Sub das Dienstkonto als Identität für den Zugriff auf Ressourcen von AWS.

Weitere Informationen zum Erstellen eines Dienstkontos, einschließlich Voraussetzungen, erforderlicher Rollen und Berechtigungen sowie Richtlinien für die Benennung, finden Sie unter Dienstkonten erstellen. Nach dem Erstellen Dienstkonto angemeldet sind, müssen Sie möglicherweise 60 Sekunden oder länger warten, bevor Sie das Dienstkonto verwenden. Dieses Verhalten tritt auf, weil Lesevorgänge Eventual Consistency haben. Es kann einige Zeit dauern, bis das neue Dienstkonto sichtbar ist.

Eindeutige ID des Dienstkontos notieren

Sie benötigen eine eindeutige ID für das Dienstkonto, um eine Rolle in der AWS-Konsole einzurichten.

1. Rufen Sie in der Google Cloud Console die Seite Dienstkonto auf.

   Zum Dienstkonto

2. Klicken Sie auf das Dienstkonto, das Sie gerade erstellt haben oder auf das Sie gerade sind. die Sie nutzen möchten.

3. Notieren Sie sich auf der Seite Dienstkontodetails die eindeutige ID.

   Sie benötigen die ID im Abschnitt Rolle in AWS mit benutzerdefinierter Vertrauensrichtlinie erstellen.

Pub/Sub-Dienstkonto die Rolle „Ersteller von Dienstkonto-Tokens“ hinzufügen

Hinweis: Pub/Sub erstellt und verwaltet für jedes Projekt ein Dienstkonto. Das Dienstkonto hat das Format service-{PROJECT_NUMBER}@gcp-sa-pubsub.iam.gserviceaccount.com. Perform nur dann ausführen, wenn das Pub/Sub- Dienstkonto hat nicht die Rolle „Pub/Sub-Dienst-Agent“ (roles/pubsub.serviceAgent)

Mit der Rolle Ersteller von Dienstkonto-Tokens (roles/iam.serviceAccountTokenCreator) können Hauptkonten kurzlebige Anmeldedaten für ein Dienstkonto erstellen. Diese Tokens oder Anmeldedaten werden verwendet, um das Dienstkonto zu imitieren.

Weitere Informationen zu Identitätswechsel für Dienstkonten.

Sie können während dieses Vorgangs auch die Rolle „Pub/Sub-Publisher“ (roles/pubsub.publisher) hinzufügen. Weitere Informationen zur Rolle und dazu, warum Sie sie hinzufügen, Siehe Pub/Sub-Publisher-Rolle zum Pub/Sub-Dienstkonto hinzufügen.

1. Öffnen Sie in der Google Cloud Console die Seite IAM.

   IAM aufrufen

2. Aktivieren Sie die Option Von Google bereitgestellte Rollenzuweisungen einbeziehen.

3. Suchen Sie nach dem Dienstkonto im Format service-{PROJECT_NUMBER}@gcp-sa-pubsub.iam.gserviceaccount.com.

4. Klicken Sie für dieses Dienstkonto auf die Schaltfläche Hauptkonto bearbeiten.

5. Klicken Sie bei Bedarf auf Weitere Rolle hinzufügen.

6. Suchen und wählen Sie die Rolle „Ersteller von Dienstkonto-Tokens“ (roles/iam.serviceAccountTokenCreator) aus.

7. Klicken Sie auf Speichern.

Richtlinie in AWS erstellen

Sie benötigen eine Richtlinie in AWS, damit sich Pub/Sub mit AWS, damit Pub/Sub Daten aus AWS Kinesis-Daten aufnehmen kann . Erstellen Sie vor dem Erstellen einer AWS-Richtlinie einen Kinesis-Datenstream mit einem registrierten Nutzer erstellt. Wir empfehlen Folgendes: um die Berechtigungen auf den jeweiligen Stream zu beschränken.

• Weitere Informationen zum Erstellen eines AWS Kinesis-Datenstreams finden Sie unter Kinesis-Datenstream

• Weitere Informationen zur AWS Kinesis Data Stream API, mit der Verbraucher registriert werden, finden Sie unter RegisterStreamConsumer.

• Weitere Methoden und Informationen zum Erstellen einer Richtlinie in AWS finden Sie unter IAM-Richtlinien erstellen.

So erstellen Sie eine Richtlinie in AWS:

1. Melden Sie sich in der AWS Management Console an und öffnen Sie die IAM-Konsole

2. Klicken Sie im Navigationsbereich der Console unter IAM auf Zugriffsverwaltung > Richtlinien.

3. Klicken Sie auf Richtlinie erstellen.

4. Wählen Sie unter Dienst auswählen die Option Kinesis aus.

5. Wählen Sie unter Aktion erlaubt Folgendes aus:

   • Liste > ListShards:

     Durch diese Aktion wird die Berechtigung zum Auflisten der Shards in einem Stream gewährt und liefert Informationen zu jedem Shard.

   • Lesen > SubscribeToShard

     Mit dieser Aktion wird die Berechtigung zum Abhören eines bestimmten Shards mit erweitertem Fan-out erteilt.

   • Lesen > DescribeStreamConsumer

     Diese Aktion gewährt die Berechtigung, die Beschreibung eines registrierten Stream-Empfängers abzurufen.

   Diese Berechtigungen decken das Lesen aus dem Stream ab. Pub/Sub unterstützt nur das Lesen aus einem Kinesis-Stream mit erweitertem Fan-Out, und zwar mithilfe der Streaming SubscribeToShard API.

6. Wenn du die Richtlinie auf einen bestimmten Stream oder Nutzer beschränken möchtest (empfohlen), gib unter Ressourcen den ARN des Nutzers und den ARN des Streams an.

7. Klicken Sie auf Weitere Berechtigungen hinzufügen.

8. Geben Sie unter Dienst auswählen STS ein und wählen Sie diese Option aus.

9. Wählen Sie unter Aktion erlaubt die Option Schreiben > AssumeRoleWithWebIdentity.

   Durch diese Aktion wird die Berechtigung zum Abrufen eines temporären Sicherheitssatzes gewährt Anmeldedaten für Pub/Sub zur Authentifizierung bei Kinesis-Daten mithilfe der Identitätsföderation streamen.

10. Klicken Sie auf Weiter.

11. Geben Sie einen Namen und eine Beschreibung für die Richtlinie ein.

12. Klicken Sie auf Richtlinie erstellen.

Rolle in AWS mit benutzerdefinierter Trust-Richtlinie erstellen

Sie müssen eine Rolle in AWS erstellen, damit sich Pub/Sub authentifizieren kann zu AWS, um Daten aus Kinesis Data Streams aufzunehmen.

So erstellen Sie eine Rolle mit einer benutzerdefinierten Vertrauensrichtlinie:

1. Melden Sie sich in der AWS Management Console an und öffnen Sie die IAM-Konsole.

2. Klicken Sie im Navigationsbereich der Console unter IAM auf Rollen.

3. Klicken Sie auf Rolle erstellen.

4. Wählen Sie unter Vertrauenswürdige Entität auswählen die Option Benutzerdefinierte Vertrauensrichtlinie aus.

5. Geben Sie im Abschnitt Benutzerdefinierte Vertrauensrichtlinie Folgendes ein oder fügen Sie es ein:

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

   Ersetzen Sie <SERVICE_ACCOUNT_UNIQUE_ID> durch die eindeutige ID von das Dienstkonto, in dem Sie sich Notieren Sie die eindeutige Dienstkonto-ID.

6. Klicken Sie auf Weiter.

7. Wählen Sie unter Berechtigungen hinzufügen die benutzerdefinierte Richtlinie aus, die die Sie gerade erstellt haben.

8. Klicken Sie auf Weiter.

9. Geben Sie einen Rollennamen und eine Beschreibung ein.

10. Klicken Sie auf Rolle erstellen.

Pub/Sub-Publisher-Rolle dem Pub/Sub-Dienstkonto hinzufügen

Sie müssen dem Pub/Sub-Dienstkonto eine Publisher-Rolle zuweisen damit Pub/Sub Inhalte im Importthema veröffentlichen kann, AWS Kinesis Data Streams

• Informationen zum Aktivieren der Veröffentlichung aus allen Themen in einem Projekt finden Sie unter Veröffentlichung aus allen Themen aktivieren.

• Informationen zum Aktivieren der Veröffentlichung aus einem bestimmten Thema (empfohlen) finden Sie unter Veröffentlichung aus einem einzelnen Thema aktivieren.

Veröffentlichung aus allen Themenbereichen aktivieren

1. Öffnen Sie in der Google Cloud Console die Seite IAM.

   IAM aufrufen

2. Aktivieren Sie die Option Von Google bereitgestellte Rollenzuweisungen einschließen.

3. Suchen Sie nach dem Dienstkonto mit folgendem Format: service-{PROJECT_NUMBER}@gcp-sa-pubsub.iam.gserviceaccount.com

4. Klicken Sie für dieses Dienstkonto auf die Schaltfläche Hauptkonto bearbeiten.

5. Klicken Sie bei Bedarf auf Weitere Rolle hinzufügen.

6. Wählen Sie die Pub/Sub-Publisher-Rolle (roles/pubsub.publisher) aus.

7. Klicken Sie auf Speichern.

Veröffentlichung über ein einzelnes Thema aktivieren

Wenn Sie die Veröffentlichungsberechtigung nur für ein bestimmtes Amazon Kinesis Data Streams-Importthema gewähren möchten, gehen Sie so vor:

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. Führen Sie den Befehl gcloud pubsub topics add-iam-policy-binding aus:

   gcloud pubsub topics add-iam-policy-binding TOPIC_ID\
      --member="serviceAccount:[e-mail]"\
      --role="roles/pubsub.publisher"

   Ersetzen Sie Folgendes:

   • TOPIC_ID ist die Topic-ID des Amazon Kinesis Data Streams-Import-Topics.

   • ist PROJECT_NUMBER die Projektnummer. So rufen Sie das Projekt auf: erhalten Sie unter Projekte identifizieren.

Dienstkonto die Rolle „Dienstkontonutzer“ hinzufügen

Die Rolle „Dienstkontonutzer“ (roles/iam.serviceAccountUser) umfasst die Berechtigung iam.serviceAccounts.actAs, mit der ein Hauptkonto ein Dienstkonto an die Aufnahmeeinstellungen des Amazon Kinesis Data Streams-Import-Themas anhängen und dieses Dienstkonto für die föderierte Identität verwenden kann.

Führen Sie diese Schritte aus:

1. Öffnen Sie in der Google Cloud Console die Seite IAM.

   IAM aufrufen

2. Für das Hauptkonto, das die Create- oder Update-Topic-Aufrufe ausgibt, Klicken Sie auf die Schaltfläche Hauptkonto bearbeiten.

3. Klicken Sie bei Bedarf auf Weitere Rolle hinzufügen.

4. Wählen Sie die Rolle „Dienstkontonutzer“ (roles/iam.serviceAccountUser) aus.

5. Klicken Sie auf Speichern.

Amazon Kinesis Data Streams-Importthema erstellen

Weitere Informationen zu Attributen, die mit einem Thema verknüpft sind, finden Sie unter Eigenschaften eines Themas.

Vergewissern Sie sich, dass Sie die folgenden Schritte ausgeführt haben:

• Föderierte Identität für den Zugriff auf Kinesis Data Streams einrichten

• Dem Pub/Sub-Dienstkonto die Rolle „Pub/Sub-Publisher“ hinzufügen

• Dem Dienstkonto die Rolle „Dienstkontonutzer“ hinzufügen

Achtung: Wenn Sie bereits Daten in AWS Kinesis Data Streams haben, müssen Sie zuerst Standardthema und Standardabo erstellen ohne die Aufnahmeeinstellungen zu aktivieren. Gehen Sie dann so vor: Konvertieren Sie das Thema in ein Importthema. So wird sichergestellt, dass alle Nachrichten vom Abo empfangen werden.

So erstellen Sie ein Amazon Kinesis Data Streams-Importthema:

Console

1. Rufen Sie in der Google Cloud Console die Seite Themen auf.

   Themen aufrufen

2. Klicken Sie auf Thema erstellen.

3. Geben Sie im Feld Topic ID (Themen-ID) eine ID für Ihr Amazon Kinesis Data Streams-Importthema ein.

   Weitere Informationen zur Benennung von Themen finden Sie in den Benennungsrichtlinien.

4. Wählen Sie Standardabo hinzufügen aus.

5. Wählen Sie Aufnahme aktivieren aus.

6. Wählen Sie als Datenaufnahmequelle Amazon Kinesis Data Streams aus.

7. Geben Sie die folgenden Informationen ein:

   • ARN des Kinesis-Streams: Der ARN für den Kinesis-Datenstream, der Daten in Pub/Sub aufnehmen möchten. Das ARN-Format lautet: arn:${Partition}:kinesis:${Region}:${Account}:stream/${StreamName}

   • ARN des Kinesis-Nutzers: Der ARN der Nutzerressource, die für den AWS Kinesis-Datenstream registriert ist. Das ARN-Format ist folgt: arn:${Partition}:kinesis:${Region}:${Account}:${StreamType}/${StreamName}/consumer/${ConsumerName}:${ConsumerCreationTimpstamp}

   • ARN der AWS-Rolle: Der ARN der AWS-Rolle. Die ARN der Rolle hat folgendes Format: arn:aws:iam:${Account}:role/${RoleName}.

   • Dienstkonto: Das Dienstkonto, das Sie unter Dienstkonto in Google Cloud erstellen erstellt haben.

8. Klicken Sie auf Thema erstellen.

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. Führen Sie den Befehl gcloud pubsub topics create aus:

   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

   Ersetzen Sie Folgendes:

   • TOPIC_ID ist die Themen-ID.

   • KINESIS_STREAM_ARN ist der ARN für die Kinesis-Datenstreams. die Sie in Pub/Sub aufnehmen möchten. Die ARN hat folgendes Format: arn:${Partition}:kinesis:${Region}:${Account}:stream/${StreamName}.

   • KINESIS_CONSUMER_ARN ist der ARN der Nutzerressource. das bei AWS Kinesis Data Streams registriert ist. Das ARN-Format ist wie folgt: arn:${Partition}:kinesis:${Region}:${Account}:${StreamType}/${StreamName}/consumer/${ConsumerName}:${ConsumerCreationTimpstamp}

   • KINESIS_ROLE_ARN ist der ARN der AWS-Rolle. Der ARN Format der Rolle: arn:aws:iam:${Account}:role/${RoleName}

   • PUBSUB_SERVICE_ACCOUNT ist das Dienstkonto, erstellt in Dienstkonto erstellen in Google Cloud.

Go

Folgen Sie der Einrichtungsanleitung für Go in der Kurzanleitung zur Verwendung von Clientbibliotheken, bevor Sie dieses Beispiel anwenden. Weitere Informationen finden Sie in der Referenzdokumentation zur Pub/Sub Go API.

Richten Sie die Standardanmeldedaten für Anwendungen ein, um sich bei Pub/Sub zu authentifizieren. Weitere Informationen finden Sie unter Authentifizierung für eine lokale Entwicklungsumgebung einrichten.

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
}

Java

Folgen Sie der Einrichtungsanleitung für Java in der Kurzanleitung zur Verwendung von Clientbibliotheken, bevor Sie dieses Beispiel anwenden. Weitere Informationen finden Sie in der Pub/Sub Java API Referenzdokumentation.

Richten Sie die Standardanmeldedaten für Anwendungen ein, um sich bei Pub/Sub zu authentifizieren. Weitere Informationen finden Sie unter Authentifizierung für eine lokale Entwicklungsumgebung einrichten.

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

Bevor Sie dieses Beispiel ausprobieren, folgen Sie der Einrichtungsanleitung für Node.js in der Pub/Sub-Schnellstart mit Clientbibliotheken. Weitere Informationen finden Sie in der Pub/Sub Node.js API Referenzdokumentation.

Richten Sie die Standardanmeldedaten für Anwendungen ein, um sich bei Pub/Sub zu authentifizieren. Weitere Informationen finden Sie unter Authentifizierung für eine lokale Entwicklungsumgebung einrichten.

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

Python

Bevor Sie dieses Beispiel ausprobieren, folgen Sie der Einrichtungsanleitung für Python in der Pub/Sub-Schnellstart mit Clientbibliotheken. Weitere Informationen finden Sie in der Referenzdokumentation zur Pub/Sub Python API.

Richten Sie die Standardanmeldedaten für Anwendungen ein, um sich bei Pub/Sub zu authentifizieren. Weitere Informationen finden Sie unter Authentifizierung für eine lokale Entwicklungsumgebung einrichten.

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

C++

Bevor Sie dieses Beispiel ausprobieren, folgen Sie der Einrichtungsanleitung für C++ in der Pub/Sub-Schnellstart mit Clientbibliotheken. Weitere Informationen finden Sie in der Referenzdokumentation zur Pub/Sub C++ API.

Richten Sie Standardanmeldedaten für Anwendungen ein, um sich bei Pub/Sub zu authentifizieren. Weitere Informationen finden Sie unter Authentifizierung für eine lokale Entwicklungsumgebung einrichten.

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

Node.js (TypeScript)

Folgen Sie der Einrichtungsanleitung für Node.js in der Pub/Sub-Kurzanleitung mit Clientbibliotheken Weitere Informationen finden Sie in der Pub/Sub Node.js API Referenzdokumentation.

Richten Sie die Standardanmeldedaten für Anwendungen ein, um sich bei Pub/Sub zu authentifizieren. Weitere Informationen finden Sie unter Authentifizierung für eine lokale Entwicklungsumgebung einrichten.

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

Weitere Informationen zu ARNs finden Sie unter Amazon-Ressourcennamen (ARNs) und IAM-IDs.

Wenn Probleme auftreten, lesen Sie den Hilfeartikel Probleme beim Importieren in Amazon Kinesis Data Streams beheben.

Amazon Kinesis Data Streams-Importthema bearbeiten

Sie können die Einstellungen für die Datenaufnahmequelle eines Amazon Kinesis Data Streams-Importthemas bearbeiten. Führen Sie diese Schritte aus:

Console

1. Rufen Sie in der Google Cloud Console die Seite Themen auf.

   Themen aufrufen

2. Klicken Sie auf das Importthema Amazon Kinesis Data Streams.

3. Klicken Sie auf der Seite mit den Themendetails auf Bearbeiten.

4. Aktualisieren Sie die Felder, die Sie ändern möchten.

5. Klicken Sie auf Aktualisieren.

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. Führen Sie den Befehl gcloud pubsub topics update mit allen im folgenden Beispiel genannten Flags aus:

     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
     

   Ersetzen Sie Folgendes:

   • TOPIC_ID ist die Themen-ID. Dieses Feld kann nicht aktualisiert werden.

   • KINESIS_STREAM_ARN ist die ARN für die Kinesis Data Streams, die Sie in Pub/Sub aufnehmen möchten. Der ARN folgendes Format: arn:${Partition}:kinesis:${Region}:${Account}:stream/${StreamName}

   • KINESIS_CONSUMER_ARN ist der ARN der Nutzerressource. das bei AWS Kinesis Data Streams registriert ist. Das ARN-Format hat folgendes Format: arn:${Partition}:kinesis:${Region}:${Account}:${StreamType}/${StreamName}/consumer/${ConsumerName}:${ConsumerCreationTimpstamp}.

   • KINESIS_ROLE_ARN ist der ARN der AWS-Rolle. Die ARN der Rolle hat folgendes Format: arn:aws:iam:${Account}:role/${RoleName}.

   • PUBSUB_SERVICE_ACCOUNT ist das Dienstkonto, das unter Dienstkonto in Google Cloud erstellen erstellt wurde.

Kontingente und Limits für Amazon Kinesis Data Streams-Importthemen

Der Publisher-Durchsatz für importierte Themen ist durch das Publish-Kontingent des Themas begrenzt. Weitere Informationen finden Sie unter Pub/Sub-Kontingente und Limits.

Nächste Schritte

• Importthema überwachen

• Wählen Sie die Art des Abos für Ihr Thema aus.

• Weitere Informationen zum Veröffentlichen von Nachrichten in einem Thema

• Themen mit der gcloud CLI, REST APIs oder Clientbibliotheken erstellen oder ändern

• Fehlerbehebung bei einem Amazon Kinesis Data Streams-Importthema

• Schritte zum Konfigurieren eines Storage Transfer-Jobs zum Übertragen von Daten von AWS S3 zu GCS