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 zur Konfiguration der Quelle 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 Streaming-Tabellen für Apache Iceberg aus.

   • Wählen Sie ein Format für die Datensatzkonvertierung (JSON oder JSON (Glue Schema Registry)). Wählen Sie auch das Glue Schema Registry für einfaches JSON.

   • Unter Lagerstandort:

     • Wählen Sie einen S3-Table-Bucket aus.

   • Unter Zieltabelle:

     • Wählen Sie den Namespace aus oder geben Sie ihn ein.

     • Geben Sie den Tabellennamen und den Spaltennamen ein und wählen Sie den Komprimierungstyp (ZSTD oder Snappy).

   • Für Streaming-Tabellen für Apache Iceberg ist die Partitionierung zeitbasiert.

   • 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. Geben Sie unter Dead-letter Warteschlange den S3-Bucket-URI für nicht verarbeitbare Datensätze an. Geben Sie optional ein Präfix für die Fehlerausgabe ein.

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 aus.

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 Streaming-Tabellen für das Apache Iceberg-Ziel zu erstellen.

aws kafka create-channel \
    --cluster-arn "arn:aws:kafka:us-east-1:123456789012:cluster/my-express-cluster/abc123" \
    --channel-name "orders-channel" \
    --topic-configuration-list '[
        {
            "TopicArn": "arn:aws:kafka:us-east-1:123456789012:topic/my-express-cluster/abc123/orders-topic",
            "RecordConverter": { "ValueConverter": "JSON" },
            "RecordSchema": { "GsrArn": "arn:aws:glue:us-east-1:123456789012:schema/my-registry/orders-schema" }
        }
    ]' \
    --iceberg-destination-configuration '{
        "ServiceExecutionRoleArn": "arn:aws:iam::123456789012:role/MSKChannelRole",
        "DataFreshnessInSeconds": 300,
        "Catalog": {
            "WarehouseLocation": "arn:aws:s3tables:us-east-1:123456789012:bucket/analytics-table-bucket"
        },
        "DeadLetterQueueS3": {
            "BucketArn": "arn:aws:s3:::my-dlq-bucket",
            "ErrorOutputPrefix": "dlq/"
        },
        "DestinationTableList": [
            {
                "DestinationDatabaseName": "analytics_db",
                "DestinationTableName": "orders_iceberg",
                "PartitionSpec": {
                    "PartitionStrategy": "TIME_HOUR",
                    "SourceList": [ { "SourceName": "timestamp" } ]
                }
            }
        ],
        "CompressionType": "ZSTD"
    }'

• RecordConverter.ValueConverterist JSON (einfaches JSON — gibt den ARN des Glue Schema Registry Schemas an, das die Daten definiert) oder JSON_SCHEMA_GSR (GSR-serialized JSON — die Schema-ID ist in jeden Datensatz eingebettet). RecordSchema.GsrArn

• Catalog ist optional. Wenn Sie angebenCatalog.CatalogArn, muss es auf den S3-Tabellen-Katalog (.../catalog/s3tablescatalog/<bucket-name>) verweisen; falls nicht angegeben, leitet Amazon MSK ihn ab. WarehouseLocation

• CompressionTypeist ZSTD (Standard) oder. SNAPPY Die Partitionierung wird PartitionStrategy: TIME_HOUR mit einer einzigen Zeitstempel-Quellspalte verwendet.

Die Antwort ist für beide Ziele dieselbe.

{
    "ChannelArn": "arn:aws:kafka:us-east-1:123456789012:channel/my-express-cluster/abc123/orders-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.