Apache Flume - Guide rapide

Qu'est-ce que Flume?

Apache Flume est un outil / service / mécanisme d'ingestion de données pour collecter l'agrégation et le transport de grandes quantités de données en continu telles que des fichiers journaux, des événements (etc ...) de diverses sources vers un magasin de données centralisé.

Flume est un outil hautement fiable, distribué et configurable. Il est principalement conçu pour copier des données en continu (données de journal) de divers serveurs Web vers HDFS.

Applications de Flume

Supposons qu'une application Web de commerce électronique souhaite analyser le comportement des clients d'une région particulière. Pour ce faire, ils doivent déplacer les données de journal disponibles dans Hadoop pour analyse. Ici, Apache Flume vient à notre secours.

Flume est utilisé pour déplacer les données de journal générées par les serveurs d'applications vers HDFS à une vitesse plus élevée.

Avantages de Flume

Voici les avantages de l'utilisation de Flume -

  • En utilisant Apache Flume, nous pouvons stocker les données dans n'importe lequel des magasins centralisés (HBase, HDFS).

  • Lorsque le taux de données entrantes dépasse le taux auquel les données peuvent être écrites vers la destination, Flume agit comme un médiateur entre les producteurs de données et les magasins centralisés et fournit un flux constant de données entre eux.

  • Flume offre la fonction de contextual routing.

  • Les transactions dans Flume sont basées sur des canaux où deux transactions (un expéditeur et un destinataire) sont gérées pour chaque message. Il garantit une livraison fiable des messages.

  • Flume est fiable, tolérant aux pannes, évolutif, gérable et personnalisable.

Caractéristiques de Flume

Certaines des caractéristiques notables de Flume sont les suivantes -

  • Flume ingère efficacement les données de journal de plusieurs serveurs Web dans un magasin centralisé (HDFS, HBase).

  • En utilisant Flume, nous pouvons obtenir les données de plusieurs serveurs immédiatement dans Hadoop.

  • En plus des fichiers journaux, Flume est également utilisé pour importer d'énormes volumes de données d'événements produits par des sites de réseaux sociaux comme Facebook et Twitter, et des sites de commerce électronique comme Amazon et Flipkart.

  • Flume prend en charge un large éventail de types de sources et de destinations.

  • Flume prend en charge les flux multi-hop, les flux de fan-in fan-out, le routage contextuel, etc.

  • Flume peut être mis à l'échelle horizontalement.

Big Data,comme nous le savons, est une collection de grands ensembles de données qui ne peuvent pas être traités à l'aide des techniques informatiques traditionnelles. Le Big Data, lorsqu'il est analysé, donne des résultats précieux.Hadoop est un framework open-source qui permet de stocker et de traiter du Big Data dans un environnement distribué sur des clusters d'ordinateurs à l'aide de modèles de programmation simples.

Streaming / Données de journal

En règle générale, la plupart des données à analyser seront produites par diverses sources de données telles que des serveurs d'applications, des sites de réseaux sociaux, des serveurs cloud et des serveurs d'entreprise. Ces données seront sous la forme delog files et events.

Log file - En général, un fichier journal est un filequi répertorie les événements / actions qui se produisent dans un système d'exploitation. Par exemple, les serveurs Web répertorient chaque demande adressée au serveur dans les fichiers journaux.

En récoltant ces données de journal, nous pouvons obtenir des informations sur -

  • les performances de l'application et localiser diverses défaillances logicielles et matérielles.
  • le comportement des utilisateurs et obtenir de meilleures informations commerciales.

La méthode traditionnelle de transfert de données dans le système HDFS consiste à utiliser le putcommander. Voyons comment utiliser leput commander.

Commande put HDFS

Le principal défi dans la gestion des données de journal consiste à déplacer ces journaux produits par plusieurs serveurs vers l'environnement Hadoop.

Hadoop File System Shellfournit des commandes pour insérer des données dans Hadoop et en lire. Vous pouvez insérer des données dans Hadoop à l'aide duput comme indiqué ci-dessous.

$ Hadoop fs –put /path of the required file  /path in HDFS where to save the file

Problème avec la commande put

Nous pouvons utiliser le putcommande de Hadoop pour transférer les données de ces sources vers HDFS. Mais, il souffre des inconvénients suivants -

  • En utilisant put commande, nous pouvons transférer only one file at a timetandis que les générateurs de données génèrent des données à un rythme beaucoup plus élevé. Comme l'analyse faite sur des données plus anciennes est moins précise, nous avons besoin d'une solution pour transférer les données en temps réel.

  • Si nous utilisons putcommande, les données doivent être empaquetées et doivent être prêtes pour le téléchargement. Étant donné que les serveurs Web génèrent des données en continu, c'est une tâche très difficile.

Ce dont nous avons besoin ici, c'est de solutions capables de surmonter les inconvénients de put commander et transférer les «données en continu» des générateurs de données vers les magasins centralisés (en particulier HDFS) avec moins de retard.

Problème avec HDFS

Dans HDFS, le fichier existe en tant qu'entrée de répertoire et la longueur du fichier sera considérée comme zéro jusqu'à ce qu'il soit fermé. Par exemple, si une source écrit des données dans HDFS et que le réseau a été interrompu au milieu de l'opération (sans fermer le fichier), les données écrites dans le fichier seront perdues.

Par conséquent, nous avons besoin d'un système fiable, configurable et maintenable pour transférer les données du journal dans HDFS.

Note- Dans le système de fichiers POSIX, chaque fois que nous accédons à un fichier (par exemple, effectuer une opération d'écriture), d'autres programmes peuvent toujours lire ce fichier (au moins la partie enregistrée du fichier). Cela est dû au fait que le fichier existe sur le disque avant sa fermeture.

Solutions disponibles

Pour envoyer des données en streaming (fichiers journaux, événements etc.,) de diverses sources vers HDFS, nous avons les outils suivants à notre disposition -

Scribe de Facebook

Scribe est un outil extrêmement populaire utilisé pour agréger et diffuser des données de journal. Il est conçu pour s'adapter à un très grand nombre de nœuds et être robuste aux pannes de réseau et de nœud.

Apache Kafka

Kafka a été développé par Apache Software Foundation. C'est un courtier de messages open source. En utilisant Kafka, nous pouvons gérer les flux avec un débit élevé et une faible latence.

Apache Flume

Apache Flume est un outil / service / mécanisme d'ingestion de données pour collecter l'agrégation et le transport de grandes quantités de données en continu telles que les données de journal, les événements (etc.) de divers serveurs Web vers un magasin de données centralisé.

Il s'agit d'un outil hautement fiable, distribué et configurable qui est principalement conçu pour transférer des données en continu de diverses sources vers HDFS.

Dans ce tutoriel, nous discuterons en détail de l'utilisation de Flume avec quelques exemples.

L'illustration suivante présente l'architecture de base de Flume. Comme le montre l'illustration,data generators (comme Facebook, Twitter) génèrent des données qui sont collectées par Flume individuel agentscourir sur eux. Par la suite, undata collector (qui est également un agent) collecte les données des agents qui sont agrégées et poussées dans un magasin centralisé tel que HDFS ou HBase.

Événement Flume

Un event est l'unité de base des données transportées à l'intérieur Flume. Il contient une charge utile de tableau d'octets qui doit être transportée de la source à la destination accompagnée d'en-têtes facultatifs. Un événement Flume typique aurait la structure suivante -

Agent de canal

Un agentest un processus démon indépendant (JVM) dans Flume. Il reçoit les données (événements) des clients ou d'autres agents et les transmet à sa prochaine destination (récepteur ou agent). Flume peut avoir plus d'un agent. Le diagramme suivant représente unFlume Agent

Comme le montre le diagramme, un agent Flume contient trois composants principaux à savoir, source, channel, et sink.

La source

UNE source est le composant d'un Agent qui reçoit les données des générateurs de données et les transfère vers un ou plusieurs canaux sous la forme d'événements Flume.

Apache Flume prend en charge plusieurs types de sources et chaque source reçoit des événements d'un générateur de données spécifié.

Example - Source Avro, source Thrift, source Twitter 1%, etc.

Canal

UNE channelest un stockage transitoire qui reçoit les événements de la source et les met en mémoire tampon jusqu'à ce qu'ils soient consommés par les puits. Il agit comme un pont entre les sources et les puits.

Ces canaux sont entièrement transactionnels et peuvent fonctionner avec n'importe quel nombre de sources et de puits.

Example - Canal JDBC, canal du système de fichiers, canal mémoire, etc.

Évier

UNE sinkstocke les données dans des magasins centralisés comme HBase et HDFS. Il consomme les données (événements) des canaux et les livre à la destination. La destination du récepteur peut être un autre agent ou les magasins centraux.

Example - évier HDFS

Note- Un agent de canal peut avoir plusieurs sources, puits et canaux. Nous avons répertorié toutes les sources, puits et canaux pris en charge dans le chapitre de configuration Flume de ce tutoriel.

Composants supplémentaires de Flume Agent

Ce que nous avons discuté ci-dessus sont les composants primitifs de l'agent. En plus de cela, nous avons quelques composants supplémentaires qui jouent un rôle essentiel dans le transfert des événements du générateur de données vers les magasins centralisés.

Intercepteurs

Les intercepteurs sont utilisés pour modifier / inspecter les événements de canal qui sont transférés entre la source et le canal.

Sélecteurs de canaux

Ceux-ci sont utilisés pour déterminer quel canal doit être choisi pour transférer les données en cas de canaux multiples. Il existe deux types de sélecteurs de canaux -

  • Default channel selectors - Ils sont également connus sous le nom de sélecteurs de canaux de réplication, ils répliquent tous les événements de chaque canal.

  • Multiplexing channel selectors - Ceux-ci déterminent le canal pour envoyer un événement basé sur l'adresse dans l'en-tête de cet événement.

Processeurs d'évier

Ceux-ci sont utilisés pour appeler un récepteur particulier du groupe sélectionné de puits. Ils sont utilisés pour créer des chemins de basculement pour vos récepteurs ou des événements d'équilibrage de charge sur plusieurs récepteurs d'un canal.

Flume est un framework utilisé pour déplacer les données de journal dans HDFS. En général, les événements et les données de journal sont générés par les serveurs de journaux et ces serveurs sont dotés d'agents Flume. Ces agents reçoivent les données des générateurs de données.

Les données de ces agents seront collectées par un nœud intermédiaire appelé Collector. Tout comme les agents, il peut y avoir plusieurs collecteurs dans Flume.

Enfin, les données de tous ces collecteurs seront agrégées et poussées vers un magasin centralisé tel que HBase ou HDFS. Le diagramme suivant explique le flux de données dans Flume.

Flux multi-sauts

Dans Flume, il peut y avoir plusieurs agents et avant d'atteindre la destination finale, un événement peut passer par plus d'un agent. Ceci est connu commemulti-hop flow.

Flux sortant

Le flux de données d'une source vers plusieurs canaux est appelé fan-out flow. Il est de deux types -

  • Replicating - Le flux de données où les données seront répliquées dans tous les canaux configurés.

  • Multiplexing - Le flux de données où les données seront envoyées à un canal sélectionné qui est mentionné dans l'en-tête de l'événement.

Flux de ventilateur

Le flux de données dans lequel les données seront transférées de plusieurs sources vers un canal est appelé fan-in flow.

Traitement des échecs

Dans Flume, pour chaque événement, deux transactions ont lieu: une chez l'expéditeur et une chez le destinataire. L'expéditeur envoie des événements au destinataire. Peu de temps après avoir reçu les données, le récepteur engage sa propre transaction et envoie un signal «reçu» à l'expéditeur. Après avoir reçu le signal, l'expéditeur valide sa transaction. (L'expéditeur ne validera pas sa transaction tant qu'il n'aura pas reçu un signal du destinataire.)

Nous avons déjà discuté de l'architecture de Flume dans le chapitre précédent. Dans ce chapitre, voyons comment télécharger et configurer Apache Flume.

Avant de continuer, vous devez disposer d'un environnement Java dans votre système. Donc, tout d'abord, assurez-vous que Java est installé sur votre système. Pour quelques exemples de ce didacticiel, nous avons utilisé Hadoop HDFS (comme récepteur). Par conséquent, nous vous recommandons d'installer Hadoop avec Java. Pour collecter plus d'informations, suivez le lien -

Installation de Flume

Tout d'abord, téléchargez la dernière version du logiciel Apache Flume depuis le site Web https://flume.apache.org/.

Étape 1

Ouvrez le site Web. Clique sur ledownloadlien sur le côté gauche de la page d'accueil. Cela vous mènera à la page de téléchargement d'Apache Flume.

Étape 2

Dans la page de téléchargement, vous pouvez voir les liens vers les fichiers binaires et sources d'Apache Flume. Cliquez sur le lien apache-flume-1.6.0-bin.tar.gz

Vous serez redirigé vers une liste de miroirs où vous pourrez démarrer votre téléchargement en cliquant sur l'un de ces miroirs. De la même manière, vous pouvez télécharger le code source d'Apache Flume en cliquant sur apache-flume-1.6.0-src.tar.gz .

Étape 3

Créez un répertoire avec le nom Flume dans le même répertoire où les répertoires d'installation de Hadoop, HBaseet d'autres logiciels ont été installés (si vous en avez déjà installé) comme indiqué ci-dessous.

$ mkdir Flume

Étape 4

Extrayez les fichiers tar téléchargés comme indiqué ci-dessous.

$ cd Downloads/ 
$ tar zxvf apache-flume-1.6.0-bin.tar.gz  
$ tar zxvf apache-flume-1.6.0-src.tar.gz

Étape 5

Déplacer le contenu d'apache-flume-1.6.0-bin.tar fichier au Flumerépertoire créé précédemment comme indiqué ci-dessous. (Supposons que nous ayons créé le répertoire Flume dans l'utilisateur local nommé Hadoop.)

$ mv apache-flume-1.6.0-bin.tar/* /home/Hadoop/Flume/

Configurer Flume

Pour configurer Flume, nous devons modifier trois fichiers à savoir, flume-env.sh, flumeconf.properties, et bash.rc.

Définition du chemin / chemin de classe

dans le .bashrc , définissez le dossier de départ, le chemin et le chemin de classe pour Flume comme indiqué ci-dessous.

dossier conf

Si vous ouvrez le conf dossier d'Apache Flume, vous aurez les quatre fichiers suivants -

  • flume-conf.properties.template,
  • flume-env.sh.template,
  • flume-env.ps1.template, et
  • log4j.properties.

Maintenant renommer

  • flume-conf.properties.template classer comme flume-conf.properties et

  • flume-env.sh.template comme flume-env.sh

flume-env.sh

Ouvert flume-env.sh fichier et définissez le JAVA_Home dans le dossier où Java a été installé sur votre système.

Vérification de l'installation

Vérifiez l'installation d'Apache Flume en parcourant le bin dossier et en tapant la commande suivante.

$ ./flume-ng

Si vous avez correctement installé Flume, vous obtiendrez une invite d'aide de Flume comme indiqué ci-dessous.

Après l'installation de Flume, nous devons le configurer à l'aide du fichier de configuration qui est un fichier de propriété Java ayant key-value pairs. Nous devons transmettre des valeurs aux clés du fichier.

Dans le fichier de configuration Flume, nous devons -

  • Nommez les composants de l'agent actuel.
  • Décrivez / Configurez la source.
  • Décrivez / configurez le puits.
  • Décrivez / configurez le canal.
  • Liez la source et le récepteur au canal.

Habituellement, nous pouvons avoir plusieurs agents dans Flume. Nous pouvons différencier chaque agent en utilisant un nom unique. Et en utilisant ce nom, nous devons configurer chaque agent.

Nommer les composants

Tout d'abord, vous devez nommer / lister les composants tels que les sources, les récepteurs et les canaux de l'agent, comme indiqué ci-dessous.

agent_name.sources = source_name 
agent_name.sinks = sink_name 
agent_name.channels = channel_name

Flume prend en charge diverses sources, puits et canaux. Ils sont répertoriés dans le tableau ci-dessous.

Sources Chaînes Les puits
  • Source Avro
  • Source d'épargne
  • Source d'exécution
  • Source JMS
  • Source du répertoire de spoule
  • Twitter 1% Firehose Source
  • Source Kafka
  • Source NetCat
  • Source du générateur de séquence
  • Sources Syslog
  • Source TCP Syslog
  • Source TCP Syslog multiport
  • Source Syslog UDP
  • Source HTTP
  • Source de stress
  • Sources héritées
  • Source héritée d'épargne
  • Source personnalisée
  • Source de scribe
  • Canal mémoire
  • Canal JDBC
  • Chaîne Kafka
  • Canal de fichier
  • Canal mémoire déversable
  • Pseudo canal de transaction
  • Évier HDFS
  • Évier de ruche
  • Évier enregistreur
  • Évier Avro
  • Évier d'épargne
  • Évier IRC
  • Évier de rouleau de fichier
  • Évier nul
  • HBaseSink
  • AsyncHBaseSink
  • MorphlineSolrSink
  • ElasticSearchSink
  • Évier de données de kite
  • Évier Kafka

Vous pouvez utiliser n'importe lequel d'entre eux. Par exemple, si vous transférez des données Twitter à l'aide d'une source Twitter via un canal mémoire vers un récepteur HDFS, et l'ID du nom de l'agentTwitterAgent, puis

TwitterAgent.sources = Twitter 
TwitterAgent.channels = MemChannel 
TwitterAgent.sinks = HDFS

Après avoir répertorié les composants de l'agent, vous devez décrire la (les) source (s), puits et canal (s) en fournissant des valeurs à leurs propriétés.

Décrire la source

Chaque source aura une liste de propriétés distincte. La propriété nommée «type» est commune à chaque source, et elle est utilisée pour spécifier le type de la source que nous utilisons.

Avec la propriété «type», il est nécessaire de fournir les valeurs de tous les required propriétés d'une source particulière pour la configurer, comme indiqué ci-dessous.

agent_name.sources. source_name.type = value 
agent_name.sources. source_name.property2 = value 
agent_name.sources. source_name.property3 = value

Par exemple, si nous considérons le twitter source, voici les propriétés auxquelles nous devons fournir des valeurs pour le configurer.

TwitterAgent.sources.Twitter.type = Twitter (type name) 
TwitterAgent.sources.Twitter.consumerKey =  
TwitterAgent.sources.Twitter.consumerSecret = 
TwitterAgent.sources.Twitter.accessToken =   
TwitterAgent.sources.Twitter.accessTokenSecret =

Décrire l'évier

Tout comme la source, chaque récepteur aura une liste de propriétés distincte. La propriété nommée «type» est commune à chaque récepteur, et elle est utilisée pour spécifier le type de récepteur que nous utilisons. En plus de la propriété «type», il est nécessaire de fournir des valeurs à tous lesrequired propriétés d'un récepteur particulier pour le configurer, comme indiqué ci-dessous.

agent_name.sinks. sink_name.type = value 
agent_name.sinks. sink_name.property2 = value 
agent_name.sinks. sink_name.property3 = value

Par exemple, si nous considérons HDFS sink, voici les propriétés auxquelles nous devons fournir des valeurs pour le configurer.

TwitterAgent.sinks.HDFS.type = hdfs (type name)  
TwitterAgent.sinks.HDFS.hdfs.path = HDFS directory’s Path to store the data

Décrire la chaîne

Flume fournit divers canaux pour transférer des données entre les sources et les puits. Par conséquent, avec les sources et les canaux, il est nécessaire de décrire le canal utilisé dans l'agent.

Pour décrire chaque canal, vous devez définir les propriétés requises, comme indiqué ci-dessous.

agent_name.channels.channel_name.type = value 
agent_name.channels.channel_name. property2 = value 
agent_name.channels.channel_name. property3 = value

Par exemple, si nous considérons memory channel, voici les propriétés auxquelles nous devons fournir des valeurs pour le configurer.

TwitterAgent.channels.MemChannel.type = memory (type name)

Lier la source et le puits au canal

Étant donné que les canaux connectent les sources et les puits, il est nécessaire de les lier tous les deux au canal, comme indiqué ci-dessous.

agent_name.sources.source_name.channels = channel_name 
agent_name.sinks.sink_name.channels = channel_name

L'exemple suivant montre comment lier les sources et les récepteurs à un canal. Ici, nous considéronstwitter source, memory channel, et HDFS sink.

TwitterAgent.sources.Twitter.channels = MemChannel
TwitterAgent.sinks.HDFS.channels = MemChannel

Démarrage d'un agent Flume

Après configuration, nous devons démarrer l'agent Flume. Cela se fait comme suit -

$ bin/flume-ng agent --conf ./conf/ -f conf/twitter.conf 
Dflume.root.logger=DEBUG,console -n TwitterAgent

où -

  • agent - Commande pour démarrer l'agent Flume

  • --conf ,-c<conf> - Utilisez le fichier de configuration dans le répertoire conf

  • -f<file> - Spécifie un chemin de fichier de configuration, s'il est manquant

  • --name, -n <name> - Nom de l'agent Twitter

  • -D property =value - Définit une valeur de propriété système Java.

En utilisant Flume, nous pouvons récupérer les données de divers services et les transporter vers des magasins centralisés (HDFS et HBase). Ce chapitre explique comment récupérer les données du service Twitter et les stocker dans HDFS à l'aide d'Apache Flume.

Comme indiqué dans Flume Architecture, un serveur Web génère des données de journal et ces données sont collectées par un agent dans Flume. Le canal met ces données en mémoire tampon dans un puits, qui les pousse finalement vers des magasins centralisés.

Dans l'exemple fourni dans ce chapitre, nous allons créer une application et en récupérer les tweets en utilisant la source Twitter expérimentale fournie par Apache Flume. Nous utiliserons le canal mémoire pour tamponner ces tweets et le récepteur HDFS pour pousser ces tweets dans le HDFS.

Pour récupérer les données Twitter, nous devrons suivre les étapes ci-dessous -

  • Créer une application Twitter
  • Installer / démarrer HDFS
  • Configurer Flume

Créer une application Twitter

Afin d'obtenir les tweets de Twitter, il est nécessaire de créer une application Twitter. Suivez les étapes ci-dessous pour créer une application Twitter.

Étape 1

Pour créer une application Twitter, cliquez sur le lien suivant https://apps.twitter.com/. Connectez-vous à votre compte Twitter. Vous aurez une fenêtre de gestion des applications Twitter dans laquelle vous pourrez créer, supprimer et gérer des applications Twitter.

Étape 2

Clique sur le Create New Appbouton. Vous serez redirigé vers une fenêtre où vous obtiendrez un formulaire de candidature dans lequel vous devrez remplir vos coordonnées afin de créer l'application. Lors du remplissage de l'adresse du site Web, indiquez le modèle d'URL complet, par exemple,http://example.com.

Étape 3

Remplissez les détails, acceptez le Developer Agreement une fois terminé, cliquez sur le Create your Twitter application buttonqui se trouve au bas de la page. Si tout se passe bien, une application sera créée avec les détails indiqués ci-dessous.

Étape 4

En dessous de keys and Access Tokens onglet en bas de page, vous pouvez observer un bouton nommé Create my access token. Cliquez dessus pour générer le jeton d'accès.

Étape 5

Enfin, cliquez sur le Test OAuthbouton qui se trouve en haut à droite de la page. Cela mènera à une page qui affiche votreConsumer key, Consumer secret, Access token, et Access token secret. Copiez ces détails. Celles-ci sont utiles pour configurer l'agent dans Flume.

Démarrage de HDFS

Puisque nous stockons les données dans HDFS, nous devons installer / vérifier Hadoop. Démarrez Hadoop et créez-y un dossier pour stocker les données Flume. Suivez les étapes ci-dessous avant de configurer Flume.

Étape 1: installer / vérifier Hadoop

Installez Hadoop . Si Hadoop est déjà installé sur votre système, vérifiez l'installation à l'aide de la commande de version Hadoop, comme indiqué ci-dessous.

$ hadoop version

Si votre système contient Hadoop, et si vous avez défini la variable de chemin, vous obtiendrez la sortie suivante -

Hadoop 2.6.0 
Subversion https://git-wip-us.apache.org/repos/asf/hadoop.git -r 
e3496499ecb8d220fba99dc5ed4c99c8f9e33bb1 
Compiled by jenkins on 2014-11-13T21:10Z 
Compiled with protoc 2.5.0 
From source with checksum 18e43357c8f927c0695f1e9522859d6a 
This command was run using /home/Hadoop/hadoop/share/hadoop/common/hadoop-common-2.6.0.jar

Étape 2: Démarrer Hadoop

Parcourez le sbin répertoire de Hadoop et démarrez yarn et Hadoop dfs (système de fichiers distribué) comme indiqué ci-dessous.

cd /$Hadoop_Home/sbin/ 
$ start-dfs.sh 
localhost: starting namenode, logging to
   /home/Hadoop/hadoop/logs/hadoop-Hadoop-namenode-localhost.localdomain.out 
localhost: starting datanode, logging to 
   /home/Hadoop/hadoop/logs/hadoop-Hadoop-datanode-localhost.localdomain.out 
Starting secondary namenodes [0.0.0.0] 
starting secondarynamenode, logging to 
   /home/Hadoop/hadoop/logs/hadoop-Hadoop-secondarynamenode-localhost.localdomain.out
  
$ start-yarn.sh 
starting yarn daemons 
starting resourcemanager, logging to 
   /home/Hadoop/hadoop/logs/yarn-Hadoop-resourcemanager-localhost.localdomain.out 
localhost: starting nodemanager, logging to 
   /home/Hadoop/hadoop/logs/yarn-Hadoop-nodemanager-localhost.localdomain.out

Étape 3: créer un répertoire dans HDFS

Dans Hadoop DFS, vous pouvez créer des répertoires à l'aide de la commande mkdir. Parcourez-le et créez un répertoire avec le nomtwitter_data dans le chemin requis comme indiqué ci-dessous.

$cd /$Hadoop_Home/bin/ 
$ hdfs dfs -mkdir hdfs://localhost:9000/user/Hadoop/twitter_data

Configurer Flume

Nous devons configurer la source, le canal et le puits en utilisant le fichier de configuration dans le confdossier. L'exemple donné dans ce chapitre utilise une source expérimentale fournie par Apache Flume nomméeTwitter 1% Firehose Canal mémoire et puits HDFS.

Twitter 1% Source Firehose

Cette source est hautement expérimentale. Il se connecte à l'échantillon de 1% Twitter Firehose à l'aide de l'API de streaming et télécharge en continu les tweets, les convertit au format Avro et envoie les événements Avro à un récepteur Flume en aval.

Nous obtiendrons cette source par défaut avec l'installation de Flume. lejar les fichiers correspondant à cette source peuvent être localisés dans le lib dossier comme indiqué ci-dessous.

Définition du chemin de classe

Met le classpath variable à la lib dossier de Flume dans Flume-env.sh fichier comme indiqué ci-dessous.

export CLASSPATH=$CLASSPATH:/FLUME_HOME/lib/*

Cette source a besoin des détails tels que Consumer key, Consumer secret, Access token, et Access token secretd'une application Twitter. Lors de la configuration de cette source, vous devez fournir des valeurs aux propriétés suivantes -

  • Channels

  • Source type : org.apache.flume.source.twitter.TwitterSource

  • consumerKey - La clé consommateur OAuth

  • consumerSecret - Secret du consommateur OAuth

  • accessToken - Jeton d'accès OAuth

  • accessTokenSecret - Secret du jeton OAuth

  • maxBatchSize- Nombre maximum de messages Twitter qui devraient être dans un lot Twitter. La valeur par défaut est 1 000 (facultatif).

  • maxBatchDurationMillis- Nombre maximum de millisecondes à attendre avant de fermer un lot. La valeur par défaut est 1 000 (facultatif).

Canal

Nous utilisons le canal mémoire. Pour configurer le canal mémoire, vous devez fournir une valeur au type du canal.

  • type- Il contient le type de chaîne. Dans notre exemple, le type estMemChannel.

  • Capacity- C'est le nombre maximum d'événements stockés dans le canal. Sa valeur par défaut est 100 (facultatif).

  • TransactionCapacity- C'est le nombre maximum d'événements que le canal accepte ou envoie. Sa valeur par défaut est 100 (facultatif).

Évier HDFS

Ce récepteur écrit des données dans le HDFS. Pour configurer ce récepteur, vous devez fournir les détails suivants.

  • Channel

  • type - hdfs

  • hdfs.path - le chemin du répertoire dans HDFS où les données doivent être stockées.

Et nous pouvons fournir des valeurs facultatives basées sur le scénario. Vous trouverez ci-dessous les propriétés facultatives du récepteur HDFS que nous configurons dans notre application.

  • fileType - C'est le format de fichier requis de notre fichier HDFS. SequenceFile, DataStream et CompressedStreamsont les trois types disponibles avec ce flux. Dans notre exemple, nous utilisons leDataStream.

  • writeFormat - Peut être texte ou inscriptible.

  • batchSize- Il s'agit du nombre d'événements écrits dans un fichier avant qu'il ne soit vidé dans le HDFS. Sa valeur par défaut est 100.

  • rollsize- C'est la taille du fichier pour déclencher un roulement. Sa valeur par défaut est 100.

  • rollCount- C'est le nombre d'événements écrits dans le fichier avant qu'il ne soit lancé. Sa valeur par défaut est 10.

Exemple - Fichier de configuration

Ci-dessous est un exemple du fichier de configuration. Copiez ce contenu et enregistrez-le soustwitter.conf dans le dossier conf de Flume.

# Naming the components on the current agent. 
TwitterAgent.sources = Twitter 
TwitterAgent.channels = MemChannel 
TwitterAgent.sinks = HDFS
  
# Describing/Configuring the source 
TwitterAgent.sources.Twitter.type = org.apache.flume.source.twitter.TwitterSource
TwitterAgent.sources.Twitter.consumerKey = Your OAuth consumer key
TwitterAgent.sources.Twitter.consumerSecret = Your OAuth consumer secret 
TwitterAgent.sources.Twitter.accessToken = Your OAuth consumer key access token 
TwitterAgent.sources.Twitter.accessTokenSecret = Your OAuth consumer key access token secret 
TwitterAgent.sources.Twitter.keywords = tutorials point,java, bigdata, mapreduce, mahout, hbase, nosql
  
# Describing/Configuring the sink 

TwitterAgent.sinks.HDFS.type = hdfs 
TwitterAgent.sinks.HDFS.hdfs.path = hdfs://localhost:9000/user/Hadoop/twitter_data/
TwitterAgent.sinks.HDFS.hdfs.fileType = DataStream 
TwitterAgent.sinks.HDFS.hdfs.writeFormat = Text 
TwitterAgent.sinks.HDFS.hdfs.batchSize = 1000
TwitterAgent.sinks.HDFS.hdfs.rollSize = 0 
TwitterAgent.sinks.HDFS.hdfs.rollCount = 10000 
 
# Describing/Configuring the channel 
TwitterAgent.channels.MemChannel.type = memory 
TwitterAgent.channels.MemChannel.capacity = 10000 
TwitterAgent.channels.MemChannel.transactionCapacity = 100
  
# Binding the source and sink to the channel 
TwitterAgent.sources.Twitter.channels = MemChannel
TwitterAgent.sinks.HDFS.channel = MemChannel

Exécution

Parcourez le répertoire de base de Flume et exécutez l'application comme indiqué ci-dessous.

$ cd $FLUME_HOME 
$ bin/flume-ng agent --conf ./conf/ -f conf/twitter.conf 
Dflume.root.logger=DEBUG,console -n TwitterAgent

Si tout se passe bien, la diffusion des tweets dans HDFS commencera. Ci-dessous, un instantané de la fenêtre d'invite de commande lors de la récupération des tweets.

Vérification de HDFS

Vous pouvez accéder à l'interface Web d'administration Hadoop à l'aide de l'URL ci-dessous.

http://localhost:50070/

Cliquez sur le menu déroulant nommé Utilitiessur le côté droit de la page. Vous pouvez voir deux options comme indiqué dans l'instantané ci-dessous.

Cliquer sur Browse the file systemet entrez le chemin du répertoire HDFS dans lequel vous avez stocké les tweets. Dans notre exemple, le chemin sera/user/Hadoop/twitter_data/. Ensuite, vous pouvez voir la liste des fichiers journaux Twitter stockés dans HDFS comme indiqué ci-dessous.

Dans le chapitre précédent, nous avons vu comment récupérer des données de la source Twitter vers HDFS. Ce chapitre explique comment récupérer des données depuisSequence generator.

Conditions préalables

Pour exécuter l'exemple fourni dans ce chapitre, vous devez installer HDFS de même que Flume. Par conséquent, vérifiez l'installation de Hadoop et démarrez le HDFS avant de continuer. (Reportez-vous au chapitre précédent pour savoir comment démarrer le HDFS).

Configurer Flume

Nous devons configurer la source, le canal et le puits en utilisant le fichier de configuration dans le confdossier. L'exemple donné dans ce chapitre utilise unsequence generator source, une memory channel, Et un HDFS sink.

Source du générateur de séquence

C'est la source qui génère les événements en continu. Il maintient un compteur qui commence à 0 et incrémente de 1. Il est utilisé à des fins de test. Lors de la configuration de cette source, vous devez fournir des valeurs aux propriétés suivantes -

  • Channels

  • Source type - seq

Canal

Nous utilisons le memorycanal. Pour configurer le canal mémoire, vous devez fournir une valeur au type du canal. Vous trouverez ci-dessous la liste des propriétés que vous devez fournir lors de la configuration du canal mémoire -

  • type- Il contient le type de chaîne. Dans notre exemple, le type est MemChannel.

  • Capacity- C'est le nombre maximum d'événements stockés dans le canal. Sa valeur par défaut est 100. (facultatif)

  • TransactionCapacity- C'est le nombre maximum d'événements que le canal accepte ou envoie. Sa valeur par défaut est 100. (facultatif).

Évier HDFS

Ce récepteur écrit des données dans le HDFS. Pour configurer ce récepteur, vous devez fournir les détails suivants.

  • Channel

  • type - hdfs

  • hdfs.path - le chemin du répertoire dans HDFS où les données doivent être stockées.

Et nous pouvons fournir des valeurs facultatives basées sur le scénario. Vous trouverez ci-dessous les propriétés facultatives du récepteur HDFS que nous configurons dans notre application.

  • fileType - C'est le format de fichier requis de notre fichier HDFS. SequenceFile, DataStream et CompressedStreamsont les trois types disponibles avec ce flux. Dans notre exemple, nous utilisons leDataStream.

  • writeFormat - Peut être texte ou inscriptible.

  • batchSize- Il s'agit du nombre d'événements écrits dans un fichier avant qu'il ne soit vidé dans le HDFS. Sa valeur par défaut est 100.

  • rollsize- C'est la taille du fichier pour déclencher un roulement. Sa valeur par défaut est 100.

  • rollCount- C'est le nombre d'événements écrits dans le fichier avant qu'il ne soit lancé. Sa valeur par défaut est 10.

Exemple - Fichier de configuration

Ci-dessous est un exemple du fichier de configuration. Copiez ce contenu et enregistrez-le sousseq_gen .conf dans le dossier conf de Flume.

# Naming the components on the current agent 

SeqGenAgent.sources = SeqSource   
SeqGenAgent.channels = MemChannel 
SeqGenAgent.sinks = HDFS 
 
# Describing/Configuring the source 
SeqGenAgent.sources.SeqSource.type = seq
  
# Describing/Configuring the sink
SeqGenAgent.sinks.HDFS.type = hdfs 
SeqGenAgent.sinks.HDFS.hdfs.path = hdfs://localhost:9000/user/Hadoop/seqgen_data/
SeqGenAgent.sinks.HDFS.hdfs.filePrefix = log 
SeqGenAgent.sinks.HDFS.hdfs.rollInterval = 0
SeqGenAgent.sinks.HDFS.hdfs.rollCount = 10000
SeqGenAgent.sinks.HDFS.hdfs.fileType = DataStream 
 
# Describing/Configuring the channel 
SeqGenAgent.channels.MemChannel.type = memory 
SeqGenAgent.channels.MemChannel.capacity = 1000 
SeqGenAgent.channels.MemChannel.transactionCapacity = 100 
 
# Binding the source and sink to the channel 
SeqGenAgent.sources.SeqSource.channels = MemChannel
SeqGenAgent.sinks.HDFS.channel = MemChannel

Exécution

Parcourez le répertoire de base de Flume et exécutez l'application comme indiqué ci-dessous.

$ cd $FLUME_HOME 
$./bin/flume-ng agent --conf $FLUME_CONF --conf-file $FLUME_CONF/seq_gen.conf 
   --name SeqGenAgent

Si tout se passe bien, la source commence à générer des numéros de séquence qui seront poussés dans le HDFS sous la forme de fichiers journaux.

Ci-dessous, un instantané de la fenêtre d'invite de commandes récupérant les données générées par le générateur de séquence dans le HDFS.

Vérification du HDFS

Vous pouvez accéder à l'interface Web d'administration Hadoop à l'aide de l'URL suivante:

http://localhost:50070/

Cliquez sur le menu déroulant nommé Utilitiessur le côté droit de la page. Vous pouvez voir deux options comme indiqué dans le diagramme ci-dessous.

Cliquer sur Browse the file system et entrez le chemin du répertoire HDFS dans lequel vous avez stocké les données générées par le générateur de séquence.

Dans notre exemple, le chemin sera /user/Hadoop/ seqgen_data /. Ensuite, vous pouvez voir la liste des fichiers journaux générés par le générateur de séquence, stockés dans le HDFS comme indiqué ci-dessous.

Vérification du contenu du fichier

Tous ces fichiers journaux contiennent des nombres au format séquentiel. Vous pouvez vérifier le contenu de ces fichiers dans le système de fichiers en utilisant lecat comme indiqué ci-dessous.

Ce chapitre prend un exemple pour expliquer comment vous pouvez générer des événements et les consigner par la suite dans la console. Pour cela, nous utilisons leNetCat source et le logger évier.

Conditions préalables

Pour exécuter l'exemple fourni dans ce chapitre, vous devez installer Flume.

Configurer Flume

Nous devons configurer la source, le canal et le puits en utilisant le fichier de configuration dans le confdossier. L'exemple donné dans ce chapitre utilise unNetCat Source, Memory channel, et un logger sink.

Source NetCat

Lors de la configuration de la source NetCat, nous devons spécifier un port lors de la configuration de la source. Maintenant, la source (source NetCat) écoute le port donné et reçoit chaque ligne que nous avons entrée dans ce port en tant qu'événement individuel et la transfère au puits via le canal spécifié.

Lors de la configuration de cette source, vous devez fournir des valeurs aux propriétés suivantes -

  • channels

  • Source type - netcat

  • bind - Nom d'hôte ou adresse IP à lier.

  • port - Numéro de port que nous voulons que la source écoute.

Canal

Nous utilisons le memorycanal. Pour configurer le canal mémoire, vous devez fournir une valeur au type du canal. Vous trouverez ci-dessous la liste des propriétés que vous devez fournir lors de la configuration du canal mémoire -

  • type- Il contient le type de chaîne. Dans notre exemple, le type estMemChannel.

  • Capacity- C'est le nombre maximum d'événements stockés dans le canal. Sa valeur par défaut est 100. (facultatif)

  • TransactionCapacity- C'est le nombre maximum d'événements que le canal accepte ou envoie. Sa valeur par défaut est 100. (facultatif).

Évier enregistreur

Ce récepteur enregistre tous les événements qui lui sont transmis. Généralement, il est utilisé à des fins de test ou de débogage. Pour configurer ce récepteur, vous devez fournir les détails suivants.

  • Channel

  • type - enregistreur

Exemple de fichier de configuration

Ci-dessous est un exemple du fichier de configuration. Copiez ce contenu et enregistrez-le sousnetcat.conf dans le dossier conf de Flume.

# Naming the components on the current agent
NetcatAgent.sources = Netcat   
NetcatAgent.channels = MemChannel 
NetcatAgent.sinks = LoggerSink  

# Describing/Configuring the source 
NetcatAgent.sources.Netcat.type = netcat 
NetcatAgent.sources.Netcat.bind = localhost
NetcatAgent.sources.Netcat.port = 56565  

# Describing/Configuring the sink 
NetcatAgent.sinks.LoggerSink.type = logger  

# Describing/Configuring the channel 
NetcatAgent.channels.MemChannel.type = memory 
NetcatAgent.channels.MemChannel.capacity = 1000 
NetcatAgent.channels.MemChannel.transactionCapacity = 100 
 
# Bind the source and sink to the channel 
NetcatAgent.sources.Netcat.channels = MemChannel
NetcatAgent.sinks. LoggerSink.channel = MemChannel

Exécution

Parcourez le répertoire de base de Flume et exécutez l'application comme indiqué ci-dessous.

$ cd $FLUME_HOME
$ ./bin/flume-ng agent --conf $FLUME_CONF --conf-file $FLUME_CONF/netcat.conf 
   --name NetcatAgent -Dflume.root.logger=INFO,console

Si tout se passe bien, la source commence à écouter le port donné. Dans ce cas, c'est56565. Vous trouverez ci-dessous un instantané de la fenêtre d'invite de commande d'une source NetCat qui a démarré et qui écoute le port 56565.

Passer des données à la source

Pour transmettre des données à la source NetCat, vous devez ouvrir le port indiqué dans le fichier de configuration. Ouvrez un terminal séparé et connectez-vous à la source (56565) à l'aide ducurlcommander. Une fois la connexion établie, vous recevrez un message "connected" comme indiqué ci-dessous.

$ curl telnet://localhost:56565 
connected

Vous pouvez maintenant saisir vos données ligne par ligne (après chaque ligne, vous devez appuyer sur Entrée). La source NetCat reçoit chaque ligne comme un événement individuel et vous recevrez un message reçu "OK».

Chaque fois que vous avez terminé de transmettre des données, vous pouvez quitter la console en appuyant sur (Ctrl+C). Ci-dessous se trouve l'instantané de la console où nous nous sommes connectés à la source à l'aide ducurl commander.

Chaque ligne saisie dans la console ci-dessus sera reçue en tant qu'événement individuel par la source. Depuis que nous avons utilisé leLogger récepteur, ces événements seront connectés à la console (console source) via le canal spécifié (canal mémoire dans ce cas).

L'instantané suivant montre la console NetCat où les événements sont enregistrés.