Object Storage Service:Diffusez les données EMR Kafka vers OSS-HDFS avec Flume

Prérequis

• Le service OSS-HDFS est activé et vous disposez des autorisations d'accès requises. Pour plus d'informations, consultez la section Activer le service OSS-HDFS.

• Vous avez créé un cluster DataLake et sélectionné le service Flume. Pour plus d'informations, consultez la section Créer un cluster.

• Vous avez créé un cluster DataFlow et sélectionné le service Kafka. Pour plus d'informations, consultez la section Créer un cluster.

Procédure

1. Configurez Flume.

   1. Accédez à la page de configuration de Flume.

      1. Connectez-vous à la console E-MapReduce.

      2. Dans la barre de navigation supérieure, sélectionnez une région et un groupe de ressources.

      3. Sur la page EMR on ECS, localisez le cluster cible et cliquez sur Services dans la colonne Actions.

      4. Sous l'onglet Services, cliquez sur Configure dans la zone du service FLUME.

   2. Définissez la taille maximale du tas Java Virtual Machine (JVM) (Xmx).

      L'écriture de données depuis Flume vers le service OSS-HDFS consomme beaucoup de mémoire. Nous vous recommandons d'augmenter la valeur Xmx pour l'agent Flume.

      1. Cliquez sur l'onglet flume-env.sh.

         Cette rubrique utilise une configuration globale. Pour configurer un nœud spécifique, sélectionnez Independent Node Configuration dans la liste déroulante de la page Configure du service FLUME.

      2. Modifiez la valeur du paramètre JAVA_OPTS.

         Par exemple, pour définir la taille maximale du tas JVM sur 1 Go, remplacez la valeur du paramètre par -Xmx1g.

      3. Cliquez sur Enregistrer.

   3. Modifiez la configuration flume-conf.properties.

      1. Cliquez sur l'onglet flume-conf.properties.

         Cette rubrique utilise une configuration globale. Pour configurer un nœud spécifique, sélectionnez Independent Node Configuration dans la liste déroulante de la page Configure du service FLUME.

      2. Dans l'éditeur flume-conf.properties, saisissez la configuration suivante.

         Remarque

         La valeur de default-agent dans l'exemple ci-dessous doit correspondre à la valeur du paramètre agent_name sur la page Configure du service FLUME.

         default-agent.sources = source1
         default-agent.sinks = k1
         default-agent.channels = c1
         
         default-agent.sources.source1.type = org.apache.flume.source.kafka.KafkaSource
         default-agent.sources.source1.channels = c1
         default-agent.sources.source1.kafka.bootstrap.servers = <kafka-host1:port1,kafka-host2:port2...>
         default-agent.sources.source1.kafka.topics = flume-test
         default-agent.sources.source1.kafka.consumer.group.id = flume-test-group
         
         default-agent.sinks.k1.type = hdfs
         default-agent.sinks.k1.hdfs.path = oss://{yourBucketName}.{yourBucketRegion}.oss-dls.aliyuncs.com/{path}
         default-agent.sinks.k1.hdfs.fileType=DataStream
         
         # Use a memory channel to buffer events
         default-agent.channels.c1.type = memory
         default-agent.channels.c1.capacity = 100
         default-agent.channels.c1.transactionCapacity = 100
         
         # Bind the source and sink to the channel
         default-agent.sources.source1.channels = c1
         default-agent.sinks.k1.channel = c1

         Paramètre	Description
         default-agent.sources.source1.kafka.bootstrap.servers	Les noms d'hôte et les numéros de port des brokers de votre cluster Kafka.
         default-agent.sinks.k1.hdfs.path	Le chemin d'accès OSS-HDFS doit respecter le format suivant : oss://{yourBucketName}.{yourBucketRegion}.oss-dls.aliyuncs.com/{path}. Par exemple : oss://flume-test.cn-hangzhou.oss-dls.aliyuncs.com/result. Composants du chemin d'accès : {yourBucketName} : Saisissez le nom du bucket pour lequel le service OSS-HDFS est activé. {yourBucketRegion} : Saisissez l'ID de région du bucket. {path} : Saisissez le nom du répertoire du service OSS-HDFS.
         default-agent.channels.c1.capacity	Le nombre maximal d'événements pouvant être stockés dans le canal. Ajustez cette valeur en fonction de votre environnement.
         default-agent.channels.c1.transactionCapacity	Le nombre maximal d'événements qu'une source peut écrire ou qu'un récepteur peut récupérer depuis le canal par transaction. Ajustez cette valeur en fonction de votre environnement.

      3. Cliquez sur Enregistrer.

2. Testez la synchronisation des données.

   1. Connectez-vous au cluster EMR DataFlow via SSH. Pour plus d'informations, consultez la section Se connecter à un cluster.

   2. Créez un topic nommé flume-test.

      kafka-topics.sh --partitions 10 --replication-factor 2 --zookeeper master-1-1:2181/emr-kafka --topic flume-test --create

   3. Générez des données de test.

      kafka-console-producer.sh --topic flume-test --broker-list master-1-1:9092

      Par exemple, saisissez abc et appuyez sur Entrée.

      Un fichier est généré dans le chemin d'accès oss://flume-test.cn-hangzhou.oss-dls.aliyuncs.com/result. Le fichier est nommé selon le format FlumeData.xxxx, avec l'horodatage actuel en millisecondes comme suffixe.