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

Prérequis

Notes d'utilisation

• Le connecteur exporte les données d'un topic 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 en savoir plus sur les limites des connecteurs, consultez la section Limites.

• Cette fonctionnalité repose sur le service Function Compute. Ce dernier offre un quota gratuit. Si vous dépassez ce quota, vous serez facturé pour toute ressource supplémentaire utilisée. Pour plus de détails sur la facturation, voir Présentation de la facturation.

• Function Compute permet d'interroger les journaux lors des invocations de fonctions. Pour plus d'informations, consultez la section 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 appartient le connecteur dans la liste déroulante Select Instance et cliquez sur Create Connector.

5. Dans l'assistant Create Connector, procédez comme suit :

   1. À l'étape Configure Basic Information, configurez les paramètres et cliquez sur Next. Le tableau suivant décrit ces 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 section Rôles liés au service.

      Parameter	Description	Example
      Name	The name of the connector. The name must meet the following requirements: The name must be up to 48 characters in length and can contain only digits, lowercase letters, and hyphens (-). The name cannot start with a hyphen (-). The name must be unique within the same ApsaraMQ for Kafka instance. A data synchronization task for a Connector must use a Group named connect-task name. If you do not manually create the Group, the system will automatically create it for you.	kafka-oss-sink
      Instance	By default, the name and ID of the instance are displayed.	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étaille ces paramètres.

      Parameter	Description	Example
      Data Source Topic	The topic from which to export data.	oss-test-input
      Consumer Thread Concurrency	The number of concurrent consumer threads for the source topic. Default value: 6. Valid values: 1 2 3 6 12	6
      Consumer Offset	The offset from which consumption starts. Valid values: Earliest Offset: Consumption starts from the first available message in the partition. Latest Offset: Consumption starts from the next message produced after the connector starts.	Earliest Offset
      VPC ID	The VPC where the data synchronization task runs. Click Configure Runtime Environment to display this parameter. By default, this parameter is set to the VPC of the ApsaraMQ for Kafka instance and requires no configuration.	vpc-bp1xpdnd3l***
      vSwitch ID	The vSwitch for the data synchronization task. Click Configure Runtime Environment to display this parameter. The vSwitch must be in the same VPC as the ApsaraMQ for Kafka instance. By default, this is the vSwitch specified when the ApsaraMQ for Kafka instance was deployed.	vsw-bp1d2jgg81***
      Failure Handling Policy	Specifies the policy for handling a message delivery failure on a topic partition. Click Configure Runtime Environment to display this parameter. Valid values: Continue Subscription: Continues to subscribe to the topic partition where the error occurred and prints an error log. Stop Subscription: Stops subscribing to the topic partition where the error occurred and prints an error log. Remarque For information on how to view logs, see Connector-related operations. For information on how to find solutions based on error codes, see Error codes.	Continue Subscription
      Resource Creation Method	The creation method for the topics and consumer group required by the connector. Click Configure Runtime Environment to display this parameter. Auto Manual	Auto
      Connector Consumer Group	The consumer group used by the connector. Click Configure Runtime Environment to display this parameter. We recommend that the name of this consumer group starts with connect-cluster.	connect-cluster-kafka-oss-sink
      Task Offset Topic	The topic for storing consumer offsets. Click Configure Runtime Environment to display this parameter. Topic: We recommend that the topic name starts with connect-offset. Number of partitions: The number of partitions must be greater than 1. Storage engine: The storage engine of the topic must be Local storage. Remarque Only Professional Edition instances support selecting Local storage when you create a topic. Standard Edition instances do not support this feature. cleanup.policy: The log cleanup policy for the topic must be compact.	connect-offset-kafka-oss-sink
      Task Configuration Topic	The topic for storing task configurations. Click Configure Runtime Environment to display this parameter. Topic: We recommend that the topic name starts with connect-config. Number of partitions: The number of partitions must be 1. Storage engine: The storage engine of the topic must be Local storage. Remarque Only Professional Edition instances support selecting Local storage when you create a topic. Standard Edition instances do not support this feature. cleanup.policy: The log cleanup policy for the topic must be compact.	connect-config-kafka-oss-sink
      Task Status Topic	The topic for storing task status. Click Configure Runtime Environment to display this parameter. Topic: We recommend that the topic name starts with connect-status. Number of partitions: We recommend that you set the number of partitions to 6. Storage engine: The storage engine of the topic must be Local storage. Remarque Only Professional Edition instances support selecting Local storage when you create a topic. Standard Edition instances do not support this feature. cleanup.policy: The log cleanup policy for the topic must be compact.	connect-status-kafka-oss-sink
      Dead-letter Queue Topic	The topic for storing exception data from the Kafka Connect framework. Click Configure Runtime Environment to display this parameter. This topic can be the same as the Error Data Topic to save topic resources. Topic: We recommend that the topic name starts with connect-error. Number of partitions: We recommend that you set the number of partitions to 6. Storage engine: The storage engine of the topic can be Local storage or Cloud storage. Remarque Only Professional Edition instances support selecting Local storage when you create a topic. Standard Edition instances do not support this feature.	connect-error-kafka-oss-sink
      Error Data Topic	The topic for storing exception data from the sink. Click Configure Runtime Environment to display this parameter. This topic can be the same as the Dead-letter Queue Topic to save topic resources. Topic: We recommend that the topic name starts with connect-error. Number of partitions: We recommend that you set the number of partitions to 6. Storage engine: The storage engine of the topic can be Local storage or Cloud storage. Remarque Only Professional Edition instances support selecting Local storage when you create a topic. Standard Edition instances do not support this feature.	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 ci-dessous présente ces paramètres.

      Parameter	Description	Example
      Bucket Name	The name of the OSS bucket.	bucket_test
      AccessKey ID	The AccessKey ID of your Alibaba Cloud account.	yourAccessKeyID
      AccessKey Secret	The AccessKey secret of your Alibaba Cloud account.	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'AccessKey ID et l'AccessKey Secret à OSS en tant que variables d'environnement lors de la création de la tâche. Une fois la création terminée, 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, localisez le connecteur que vous avez créé, puis cliquez sur Deploy dans la colonne Actions.

Envoyer des messages

Envoyez un message au topic source dans ApsaraMQ for Kafka pour vérifier que les données sont bien 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, entrez 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 entrez l'ID de partition dans la zone de texte Partition ID. Par exemple, 0. Pour interroger l'ID de partition, consultez la section Afficher l'état des partitions.

        • 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 ainsi qu'un type de connexion pour envoyer les messages.

Vérifier les résultats

Après avoir envoyé un message au topic source dans ApsaraMQ for Kafka, vérifiez le gestionnaire de fichiers OSS pour confirmer l'exportation des données. Pour plus d'informations, consultez la section Présentation des fichiers.

Si de nouveaux objets sont générés dans le bucket OSS, cela signifie que les données ont été 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 en fonction de vos besoins.