Einen Channel erstellen

Verwendung der AWS Management Console

1. Öffnen Sie die Amazon MSK-Konsole unter? https://console.aws.amazon.com/msk/home region=us-east-1#/home/.

2. Klicken Sie im Navigationsbereich auf Cluster.

3. Wählen Sie den Namen Ihres Amazon MSK Provisioned-Clusters mit Express-Brokern.

4. Wählen Sie den Tab Channel und dann Create Channel aus.

5. Gehen Sie im Schritt Quellkonfiguration wie folgt vor:

   • Kanalname: Geben Sie den Namen des Kanals ein, der erstellt werden soll.

   • Thema: Wählen Sie ein vorhandenes Thema aus. Erstelle ein neues Thema, falls noch keines existiert.

6. Fahren Sie mit dem Schritt zur Zielkonfiguration fort.

7. Wählen Sie unter Zieltyp die Option Datenübermittlung an Allzweck-S3 aus.

   • Wählen Sie das Datensatzformat (JSON, Byte-Array oder String) aus.

   • Unter Zieleinstellungen:

     • Wählen Sie Girokonto oder Kreuzkonto aus. Wenn Kontoübergreifendes Konto ausgewählt wurde, müssen Sie möglicherweise eine zusätzliche Konto-ID eingeben.

     • Wählen Sie einen S3-Bucket-URI für den Liefer-Bucket aus, oder geben Sie ihn ein.

     • Geben Sie ein optionales S3-Bucket-Präfix und eine Ausgabeschlüsselvorlage ein.

     • Wählen Sie die Speicherklasse und den Komprimierungstyp (NONE, GZIP oder ZSTD).

   • Stellen Sie die Aktualität der Daten ein (5—15 Minuten).

8. Fahren Sie mit dem Schritt Zusätzliche Konfiguration fort.

9. Wählen Sie unter Verschlüsselungskonfiguration optional einen vom Kunden verwalteten Schlüssel aus, um Ihre Verschlüsselungseinstellungen anzupassen.

10. Wählen oder erstellen Sie unter Service Access eine IAM-Rolle mit den erforderlichen Berechtigungen.

11. Wählen Sie unter Dead-letter Warteschlange die Option Current Account oder Cross Account aus und geben Sie den S3-Bucket-URI für nicht verarbeitbare Datensätze an. Geben Sie optional ein Präfix für die Fehlerausgabe ein. Wenn Cross-Konto ausgewählt wurde, müssen Sie möglicherweise eine zusätzliche Konto-ID eingeben.

12. (Optional) Wählen Sie unter Protokollzustellung eine CloudWatch Amazon-Protokollgruppe, einen Amazon S3-Bucket oder einen Amazon Data Firehose-Stream aus, um Protokolle zur Überwachung und Fehlerbehebung zu senden.

13. (Optional) Fügen Sie Tags hinzu.

14. Fahren Sie mit dem Schritt „Überprüfen“ fort.

15. Überprüfe und wähle Kanal erstellen.

Der Channel wechselt in den Status Creating und wechselt zu Aktiv, sobald die Bereitstellungspipeline vollständig bereitgestellt ist.

Verwendung der AWS CLI

Das Quellthema ist in TopicConfigurationList (genau ein Thema) festgelegt. Das Ziel ist entweder IcebergDestinationConfiguration oder S3DestinationConfiguration (schließt sich gegenseitig aus) festgelegt, und jede Zielkonfiguration enthält ihre eigeneServiceExecutionRoleArn, DataFreshnessInSeconds (300—900, Standard 600), und. DeadLetterQueueS3

Verwenden Sie den folgenden Befehl, um einen Channel mit einem Amazon S3-Allzweck-Bucket-Ziel zu erstellen.

aws kafka create-channel \
    --cluster-arn "arn:aws:kafka:us-east-1:123456789012:cluster/my-express-cluster/abc123" \
    --channel-name "orders-s3-channel" \
    --topic-configuration-list '[
        {
            "TopicArn": "arn:aws:kafka:us-east-1:123456789012:topic/my-express-cluster/abc123/orders-topic",
            "RecordConverter": { "ValueConverter": "JSON" }
        }
    ]' \
    --s3-destination-configuration '{
        "ServiceExecutionRoleArn": "arn:aws:iam::123456789012:role/MSKChannelRole",
        "DataFreshnessInSeconds": 300,
        "DeadLetterQueueS3": {
            "BucketArn": "arn:aws:s3:::my-dlq-bucket",
            "ErrorOutputPrefix": "dlq/"
        },
        "Storage": {
            "BucketArn": "arn:aws:s3:::my-delivery-bucket",
            "OutputPrefix": "expresslink/",
            "OutputKeyTemplate": "!{channel-id}/!{topic-name}/year=!{yyyy}/month=!{MM}/day=!{dd}/hour=!{HH}/!{topic-name}+!{partition-id}+!{kafka-offset}",
            "StorageClass": "STANDARD",
            "CompressionType": "GZIP"
        }
    }'

Für Amazon S3-Buckets für allgemeine Zwecke RecordConverter.ValueConverter kann dies oder sein JSON STRING (keine Schemaregistrierung erforderlich). BYTE_ARRAY InStorage, CompressionType kann sein NONEGZIP, oderZSTD, und StorageClass kann STANDARDINTELLIGENT_TIERING, oder GLACIER_IR sein.

Die Antwort ist für beide Ziele dieselbe.

{
    "ChannelArn": "arn:aws:kafka:us-east-1:123456789012:channel/my-express-cluster/abc123/orders-s3-channel",
    "ClusterOperationArn": "arn:aws:kafka:us-east-1:123456789012:cluster-operation/my-express-cluster/abc123/..."
}

Anmerkung

Verwenden Sie denClusterOperationArn, um den Status des asynchronen Vorgangs und etwaige Fehlermeldungen zu verfolgen. Der Channel beginnt CREATING und wechselt zu ihm, ACTIVE wenn die Bereitstellung abgeschlossen ist.

API-Referenz — siehe CreateChannel in der Amazon MSK-API-Referenz.