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. Sous Type de destination, sélectionnez Livraison de données vers S3 à usage général.

   • Sélectionnez le format d'enregistrement (JSON, tableau d'octets ou chaîne).

   • Dans Paramètres de destination  :

     • Sélectionnez Compte courant ou Compte croisé. Si le compte Cross a été sélectionné, vous devrez peut-être saisir un identifiant de compte supplémentaire.

     • Sélectionnez ou entrez un URI de compartiment S3 pour le compartiment de distribution.

     • Entrez un préfixe de compartiment S3 et un modèle de clé de sortie facultatifs.

     • Choisissez la classe de stockage et le type de compression (NONE, GZIP ou ZSTD).

   • 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 un canal avec une destination de bucket à usage général Amazon S3, 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-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"
        }
    }'

Pour les compartiments à usage général Amazon S3, cela RecordConverter.ValueConverter peut être JSONBYTE_ARRAY, ou STRING (aucun registre de schéma n'est requis). DansStorage, CompressionType peut être NONEGZIP, ouZSTD, et StorageClass peut être STANDARDINTELLIGENT_TIERING, ouGLACIER_IR.

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