Création d'un canal

Utilisation de AWS Console de gestion

1. Ouvrez la console Amazon MSK à https://console.aws.amazon.com/msk/home l'adresse ? region=us-east-1#/accueil/.

2. Dans le panneau de navigation, choisissez Clusters.

3. Choisissez le nom de votre cluster Amazon MSK Provisioned auprès d'Express Brokers.

4. Choisissez l'onglet Chaîne, puis choisissez Créer une chaîne.

5. Dans l'étape de configuration de la source :

   • Nom de la chaîne : entrez le nom de la chaîne à créer.

   • Sujet : sélectionnez un sujet existant. Créez un nouveau sujet s'il n'en existe pas encore.

6. Passez à l'étape de configuration de la destination.

7. Dans Type de destination, sélectionnez les tables de streaming pour Apache Iceberg.

   • Sélectionnez un format de conversion d'enregistrement (JSON ou JSON (Glue Schema Registry)). Choisissez également le schéma Glue Schema Registry pour le JSON brut.

   • Sous Emplacement de l'entrepôt  :

     • Sélectionnez Compte courant ou Compte croisé.

     • Sélectionnez un compartiment S3 Table ou entrez l'ARN d'un compartiment S3 Table si Cross account a été sélectionné.

   • Dans le tableau des destinations  :

     • Sélectionnez ou entrez l'espace de noms.

     • Entrez le nom de la table et le nom de la colonne, puis choisissez le type de compression (ZSTD ou Snappy).

   • Pour les tables de streaming pour Apache Iceberg, le partitionnement est basé sur le temps.

   • Réglez la fraîcheur des données (5 à 15 minutes).

8. Passez à l'étape de configuration supplémentaire.

9. Dans Configuration du chiffrement, sélectionnez éventuellement une clé gérée par le client pour personnaliser vos paramètres de chiffrement.

10. Sous Accès au service, sélectionnez ou créez un rôle IAM avec les autorisations requises.

11. Dans la Dead-letter file d'attente, sélectionnez Compte courant ou Compte croisé et spécifiez l'URI du compartiment S3 pour les enregistrements non traitables. Entrez éventuellement un préfixe de sortie d'erreur. Si le compte Cross a été sélectionné, vous devrez peut-être saisir un identifiant de compte supplémentaire.

12. (Facultatif) Dans Log Delivery, sélectionnez un groupe de CloudWatch journaux Amazon, un compartiment Amazon S3 ou un flux Amazon Data Firehose pour envoyer des journaux à des fins de surveillance et de dépannage.

13. (Facultatif) Ajoutez des balises.

14. Passez à l'étape Révision.

15. Passez en revue et choisissez Créer une chaîne.

Le canal passe à l'état Création et passe à Actif une fois que le pipeline de distribution est entièrement provisionné.

Utilisation de AWS INTERFACE DE LIGNE DE COMMANDE (CLI)

La rubrique source est définie dans TopicConfigurationList (exactement une rubrique). La destination est définie dans l'un IcebergDestinationConfiguration ou l'autre S3DestinationConfiguration (s'excluant mutuellement), et chaque configuration de destination contient les ServiceExecutionRoleArn siennes DataFreshnessInSeconds (300—900, par défaut 600), et. DeadLetterQueueS3

Pour créer une chaîne avec des tables de streaming pour la destination Apache Iceberg, utilisez la commande suivante.

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.ValueConverterest JSON (JSON brut — fournissezRecordSchema.GsrArn, l'ARN du schéma du registre des schémas Glue qui définit les données) ou JSON_SCHEMA_GSR (GSR-serialized JSON — l'ID du schéma est intégré dans chaque enregistrement).

• Le paramètre Catalog est facultatif. Si vous le fournissezCatalog.CatalogArn, il doit faire référence au catalogue S3 Tables (.../catalog/s3tablescatalog/<bucket-name>) ; en cas d'omission, Amazon MSK le dérive. WarehouseLocation

• CompressionTypeest ZSTD (par défaut) ouSNAPPY. Le partitionnement s'utilise PartitionStrategy: TIME_HOUR avec une seule colonne source d'horodatage.

La réponse est la même pour les deux destinations.

{
    "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/..."
}

Note

Utilisez le ClusterOperationArn pour suivre l'état de l'opération asynchrone et tout message d'erreur. Le canal démarre CREATING et passe à la fin ACTIVE du provisionnement.

Référence d'API  : voir CreateChannel dans la référence d'API Amazon MSK.