ApsaraMQ for Kafka:Créer un connecteur de réception OSS

Prérequis

Notes d'utilisation

• Le connecteur exporte les données d'une rubrique source d'une instance ApsaraMQ for Kafka vers Function Compute, puis de Function Compute vers Object Storage Service (OSS). L'instance ApsaraMQ for Kafka, le service Function Compute et le bucket OSS doivent se trouver dans la même région. Pour plus d'informations sur les limites des connecteurs, consultez la section Limites.

• Cette fonctionnalité repose sur le service Function Compute. Function Compute propose un quota gratuit. Si vous dépassez ce quota, l'utilisation des ressources supplémentaires vous sera facturée. Pour plus d'informations sur la facturation, consultez la page Présentation de la facturation.

• Function Compute permet d'interroger les journaux liés aux invocations de fonctions. Pour plus d'informations, consultez la page Configurer les journaux.

• Lors de l'exportation des messages, ApsaraMQ for Kafka les sérialise en utilisant l'encodage de chaîne UTF-8. Les formats de données binaires ne sont pas pris en charge.

Créer et déployer un connecteur de réception OSS

1. Connectez-vous à la console ApsaraMQ for Kafka.

2. Sur la page Overview, sélectionnez une région dans la section Resource Distribution.

3. Dans le volet de navigation de gauche, cliquez sur Connectors.

4. Sur la page Connectors, sélectionnez l'instance à laquelle le connecteur appartient dans la liste déroulante Select Instance et cliquez sur Create Connector.

5. Dans l'assistant Create Connector, suivez les étapes ci-dessous :

   1. À l'étape Configure Basic Information, configurez les paramètres et cliquez sur Next. Le tableau suivant décrit les paramètres.

      Important

      ApsaraMQ for Kafka sélectionne automatiquement l'option Authorize to Create Service Linked Role.

      • Si aucun rôle lié au service n'a été créé, ApsaraMQ for Kafka en crée un automatiquement pour vous. Ce rôle permet à ApsaraMQ for Kafka d'exporter des données vers OSS.

      • Si un rôle lié au service existe déjà, ApsaraMQ for Kafka n'en crée pas de nouveau.

      Pour plus d'informations sur ce rôle lié au service, consultez la page Rôles liés au service.

      Paramètre	Description	Exemple
      Name	Nom du connecteur. Le nom doit respecter les exigences suivantes : Le nom peut contenir jusqu'à 48 caractères et uniquement des chiffres, des lettres minuscules et des traits d'union (-). Il ne peut pas commencer par un trait d'union (-). Le nom doit être unique au sein de la même instance ApsaraMQ for Kafka. Une tâche de synchronisation des données pour un Connecteur doit utiliser un Group nommé connect-nom de la tâche. Si vous ne créez pas manuellement le Group, le système le créera automatiquement pour vous.	kafka-oss-sink
      Instance	Par défaut, le nom et l'ID de l'instance s'affichent.	demo alikafka_post-cn-st21p8vj****

   2. À l'étape Configure Source Service, sélectionnez Message Queue for Apache Kafka comme service source, configurez les paramètres, puis cliquez sur Next. Le tableau suivant décrit les paramètres.

      Paramètre	Description	Exemple
      Data Source Topic	Rubrique à partir de laquelle exporter les données.	oss-test-input
      Consumer Thread Concurrency	Nombre de threads de consommateur simultanés pour la rubrique source. Valeur par défaut : 6. Valeurs valides : 1 2 3 6 12	6
      Consumer Offset	Offset à partir duquel la consommation commence. Valeurs valides : Earliest Offset : La consommation commence à partir du premier message disponible dans la partition. Latest Offset : La consommation commence à partir du prochain message produit après le démarrage du connecteur.	Earliest Offset
      VPC ID	VPC dans lequel s'exécute la tâche de synchronisation des données. Cliquez sur Configure Runtime Environment pour afficher ce paramètre. Par défaut, ce paramètre est défini sur le VPC de l'instance ApsaraMQ for Kafka et ne nécessite aucune configuration.	vpc-bp1xpdnd3l***
      vSwitch ID	vSwitch pour la tâche de synchronisation des données. Cliquez sur Configure Runtime Environment pour afficher ce paramètre. Le vSwitch doit se trouver dans le même VPC que l'instance ApsaraMQ for Kafka. Par défaut, il s'agit du vSwitch spécifié lors du déploiement de l'instance ApsaraMQ for Kafka.	vsw-bp1d2jgg81***
      Failure Handling Policy	Spécifie la politique de gestion d'un échec de livraison de message sur une partition de rubrique. Cliquez sur Configure Runtime Environment pour afficher ce paramètre. Valeurs valides : Continue Subscription : Continue à s'abonner à la partition de rubrique où l'erreur s'est produite et imprime un journal d'erreurs. Stop Subscription : Arrête l'abonnement à la partition de rubrique où l'erreur s'est produite et imprime un journal d'erreurs. Remarque Pour savoir comment consulter les journaux, consultez la page Opérations liées aux connecteurs. Pour savoir comment trouver des solutions basées sur les codes d'erreur, consultez la section Codes d'erreur.	Continue Subscription
      Resource Creation Method	Méthode de création des rubriques et du groupe de consommateurs requis par le connecteur. Cliquez sur Configure Runtime Environment pour afficher ce paramètre. Auto Manual	Auto
      Connector Consumer Group	Groupe de consommateurs utilisé par le connecteur. Cliquez sur Configure Runtime Environment pour afficher ce paramètre. Nous vous recommandons de faire commencer le nom de ce groupe de consommateurs par connect-cluster.	connect-cluster-kafka-oss-sink
      Task Offset Topic	Rubrique permettant de stocker les offsets des consommateurs. Cliquez sur Configure Runtime Environment pour afficher ce paramètre. Rubrique : Nous vous recommandons de faire commencer le nom de la rubrique par connect-offset. Nombre de partitions : Le nombre de partitions doit être supérieur à 1. Moteur de stockage : Le moteur de stockage de la rubrique doit être Local storage. Remarque Seules les instances Professional Edition permettent de sélectionner Local storage lors de la création d'une rubrique. Les instances Standard Edition ne prennent pas en charge cette fonctionnalité. cleanup.policy : La politique de nettoyage des journaux pour la rubrique doit être compact.	connect-offset-kafka-oss-sink
      Task Configuration Topic	Rubrique permettant de stocker les configurations des tâches. Cliquez sur Configure Runtime Environment pour afficher ce paramètre. Rubrique : Nous vous recommandons de faire commencer le nom de la rubrique par connect-config. Nombre de partitions : Le nombre de partitions doit être égal à 1. Moteur de stockage : Le moteur de stockage de la rubrique doit être Local storage. Remarque Seules les instances Professional Edition permettent de sélectionner Local storage lors de la création d'une rubrique. Les instances Standard Edition ne prennent pas en charge cette fonctionnalité. cleanup.policy : La politique de nettoyage des journaux pour la rubrique doit être compact.	connect-config-kafka-oss-sink
      Task Status Topic	Rubrique permettant de stocker l'état des tâches. Cliquez sur Configure Runtime Environment pour afficher ce paramètre. Rubrique : Nous vous recommandons de faire commencer le nom de la rubrique par connect-status. Nombre de partitions : Nous vous recommandons de définir le nombre de partitions sur 6. Moteur de stockage : Le moteur de stockage de la rubrique doit être Local storage. Remarque Seules les instances Professional Edition permettent de sélectionner Local storage lors de la création d'une rubrique. Les instances Standard Edition ne prennent pas en charge cette fonctionnalité. cleanup.policy : La politique de nettoyage des journaux pour la rubrique doit être compact.	connect-status-kafka-oss-sink
      Dead-letter Queue Topic	Rubrique permettant de stocker les données d'exception du framework Kafka Connect. Cliquez sur Configure Runtime Environment pour afficher ce paramètre. Cette rubrique peut être identique à la rubrique Error Data Topic afin d'économiser des ressources de rubrique. Rubrique : Nous vous recommandons de faire commencer le nom de la rubrique par connect-error. Nombre de partitions : Nous vous recommandons de définir le nombre de partitions sur 6. Moteur de stockage : Le moteur de stockage de la rubrique peut être Local storage ou Cloud storage. Remarque Seules les instances Professional Edition permettent de sélectionner Local storage lors de la création d'une rubrique. Les instances Standard Edition ne prennent pas en charge cette fonctionnalité.	connect-error-kafka-oss-sink
      Error Data Topic	Rubrique permettant de stocker les données d'exception du récepteur (sink). Cliquez sur Configure Runtime Environment pour afficher ce paramètre. Cette rubrique peut être identique à la rubrique Dead-letter Queue Topic afin d'économiser des ressources de rubrique. Rubrique : Nous vous recommandons de faire commencer le nom de la rubrique par connect-error. Nombre de partitions : Nous vous recommandons de définir le nombre de partitions sur 6. Moteur de stockage : Le moteur de stockage de la rubrique peut être Local storage ou Cloud storage. Remarque Seules les instances Professional Edition permettent de sélectionner Local storage lors de la création d'une rubrique. Les instances Standard Edition ne prennent pas en charge cette fonctionnalité.	connect-error-kafka-oss-sink

   3. À l'étape Configure Destination Service, sélectionnez Object Storage Service comme service de destination, définissez les paramètres, puis cliquez sur Create. Le tableau suivant décrit les paramètres.

      Paramètre	Description	Exemple
      Bucket Name	Nom du bucket OSS.	bucket_test
      AccessKey ID	ID AccessKey de votre compte Alibaba Cloud.	yourAccessKeyID
      AccessKey Secret	Clé secrète AccessKey de votre compte Alibaba Cloud.	yourAccessKeySecret

      Assurez-vous que votre compte Alibaba Cloud dispose des autorisations suivantes, conformément au principe du moindre privilège :

      {
          "Version": "1",
          "Statement": [
              {
                  "Action": [
                      "oss:GetObject",
                      "oss:PutObject"
                  ],
                  "Resource": "*",
                  "Effect": "Allow"
              }
          ]
      }

      Remarque

      ApsaraMQ for Kafka transmet l'ID AccessKey et la clé secrète AccessKey à OSS en tant que variables d'environnement lors de la création de la tâche. Après la création, ApsaraMQ for Kafka ne stocke pas ces clés.

      Une fois le connecteur créé, vous pouvez le consulter sur la page Connectors.

6. Accédez à la page Connectors, recherchez le connecteur que vous avez créé, puis cliquez sur Deploy dans la colonne Actions.

Envoyer des messages

Envoyez un message à la rubrique source dans ApsaraMQ for Kafka pour vérifier que les données sont exportées vers OSS.

1. Sur la page Connectors, recherchez le connecteur cible et cliquez sur Test dans la colonne Actions.

2. Dans le panneau Send Message, envoyez un message de test.

   • Définissez Sending Method sur Console.

     1. Dans la zone de texte Message Key, saisissez la clé du message. Par exemple, demo.

     2. Dans la zone de texte Message Content, saisissez le contenu du message de test. Par exemple, {"key": "test"}.

     3. Définissez Send to Specified Partition pour indiquer si le message doit être envoyé à une partition spécifique.

        • Cliquez sur Yes et saisissez l'ID de partition dans la zone de texte Partition ID. Par exemple, 0. Pour interroger l'ID de partition, consultez la page Afficher l'état de la partition.

        • Cliquez sur No pour ne pas spécifier de partition.

   • Définissez Sending Method sur Docker. Exécutez la commande Docker dans la section Run the Docker container to produce a sample message pour envoyer le message.

   • Définissez Sending Method sur SDK. Sélectionnez un SDK pour le langage ou le framework requis et un type de connexion pour envoyer des messages.

Vérifier les résultats

Après avoir envoyé un message à la rubrique source dans ApsaraMQ for Kafka, vérifiez le gestionnaire de fichiers OSS pour confirmer que les données ont été exportées. Pour plus d'informations, consultez la page Présentation des fichiers.

Si de nouveaux objets sont générés dans le bucket OSS, cela signifie que les données sont synchronisées avec OSS.

[
    {
        "key":"123",
        "offset":4,
        "overflowFlag":true,
        "partition":0,
        "timestamp":1603779578478,
        "topic":"Test",
        "value":"1",
        "valueSize":272687
    }
]

Opérations connexes

Vous pouvez configurer les ressources Function Compute requises par le connecteur de réception OSS selon vos besoins.