Apache Flume — Введение

Apache Flume — это инструмент, позволяющий собирать и обрабатывать большие объемы потоковых данных, таких как файлы журналов, события (и т. Д.), Из различных источников в централизованное хранилище данных.

Flume — это высоконадежный, распределенный и настраиваемый инструмент. Он предназначен главным образом для копирования потоковых данных (данных журнала) с различных веб-серверов в HDFS.

Применение Flume

Предположим, что веб-приложение электронной коммерции хочет проанализировать поведение клиентов в конкретном регионе. Для этого им необходимо переместить доступные данные журнала в Hadoop для анализа. Здесь Apache Flume приходит нам на помощь.

Flume используется для перемещения данных журнала, генерируемых серверами приложений, в HDFS с более высокой скоростью.

Преимущества Flume

Вот преимущества использования Flume —

• Используя Apache Flume, мы можем хранить данные в любом из централизованных хранилищ (HBase, HDFS).

• Когда скорость входящих данных превышает скорость, с которой данные могут быть записаны в место назначения, Flume выступает в качестве посредника между производителями данных и централизованными хранилищами и обеспечивает устойчивый поток данных между ними.

• Flume предоставляет возможность контекстной маршрутизации .

• Транзакции в Flume основаны на каналах, где для каждого сообщения поддерживаются две транзакции (один отправитель и один получатель). Это гарантирует надежную доставку сообщений.

• Flume является надежным, отказоустойчивым, масштабируемым, управляемым и настраиваемым.

Используя Apache Flume, мы можем хранить данные в любом из централизованных хранилищ (HBase, HDFS).

Когда скорость входящих данных превышает скорость, с которой данные могут быть записаны в место назначения, Flume выступает в качестве посредника между производителями данных и централизованными хранилищами и обеспечивает устойчивый поток данных между ними.

Flume предоставляет возможность контекстной маршрутизации .

Транзакции в Flume основаны на каналах, где для каждого сообщения поддерживаются две транзакции (один отправитель и один получатель). Это гарантирует надежную доставку сообщений.

Flume является надежным, отказоустойчивым, масштабируемым, управляемым и настраиваемым.

Особенности Flume

Некоторые из примечательных особенностей Flume следующие:

• Flume эффективно загружает данные журнала с нескольких веб-серверов в централизованное хранилище (HDFS, HBase).

• Используя Flume, мы можем сразу получить данные с нескольких серверов в Hadoop.

• Вместе с файлами журналов Flume также используется для импорта огромных объемов данных о событиях, создаваемых социальными сетями, такими как Facebook и Twitter, а также сайтами электронной коммерции, такими как Amazon и Flipkart.

• Flume поддерживает большой набор типов источников и направлений.

• Flume поддерживает многоскачковые потоки, потоки разветвления, контекстную маршрутизацию и т. Д.

• Flume можно масштабировать по горизонтали.

Flume эффективно загружает данные журнала с нескольких веб-серверов в централизованное хранилище (HDFS, HBase).

Используя Flume, мы можем сразу получить данные с нескольких серверов в Hadoop.

Вместе с файлами журналов Flume также используется для импорта огромных объемов данных о событиях, создаваемых социальными сетями, такими как Facebook и Twitter, а также сайтами электронной коммерции, такими как Amazon и Flipkart.

Flume поддерживает большой набор типов источников и направлений.

Flume поддерживает многоскачковые потоки, потоки разветвления, контекстную маршрутизацию и т. Д.

Flume можно масштабировать по горизонтали.

Apache Flume — передача данных в Hadoop

Большие данные, как мы знаем, представляют собой набор больших наборов данных, которые не могут быть обработаны с использованием традиционных вычислительных технологий. Большие данные при анализе дают ценные результаты. Hadoop — это платформа с открытым исходным кодом, которая позволяет хранить и обрабатывать большие данные в распределенной среде на кластерах компьютеров с использованием простых моделей программирования.

Потоковая передача / регистрация данных

Как правило, большая часть данных, подлежащих анализу, будет создаваться различными источниками данных, такими как серверы приложений, сайты социальных сетей, облачные серверы и корпоративные серверы. Эти данные будут в виде файлов журналов и событий .

Файл журнала. Как правило, файл журнала представляет собой файл, в котором перечислены события / действия, которые происходят в операционной системе. Например, веб-серверы перечисляют каждый запрос к серверу в файлах журнала.

При сборе таких данных журнала мы можем получить информацию о —

• производительность приложения и найти различные программные и аппаратные сбои.
• поведение пользователя и получить лучшее понимание бизнеса.

Традиционный метод передачи данных в систему HDFS — использование команды put . Давайте посмотрим, как использовать команду put .

HDFS поставил Команду

Основная проблема при обработке данных журналов заключается в перемещении этих журналов, созданных несколькими серверами, в среду Hadoop.

Оболочка файловой системы Hadoop предоставляет команды для вставки данных в Hadoop и чтения из них. Вы можете вставить данные в Hadoop с помощью команды put, как показано ниже.

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

Проблема с положенной Командой

Мы можем использовать команду put Hadoop для передачи данных из этих источников в HDFS. Но он страдает от следующих недостатков —

• Используя команду put , мы можем передавать только один файл за раз, в то время как генераторы данных генерируют данные с гораздо большей скоростью. Поскольку анализ, выполненный на более старых данных, менее точен, у нас должно быть решение для передачи данных в режиме реального времени.

• Если мы используем команду put , данные должны быть упакованы и должны быть готовы к загрузке. Поскольку веб-серверы генерируют данные непрерывно, это очень сложная задача.

Используя команду put , мы можем передавать только один файл за раз, в то время как генераторы данных генерируют данные с гораздо большей скоростью. Поскольку анализ, выполненный на более старых данных, менее точен, у нас должно быть решение для передачи данных в режиме реального времени.

Если мы используем команду put , данные должны быть упакованы и должны быть готовы к загрузке. Поскольку веб-серверы генерируют данные непрерывно, это очень сложная задача.

Здесь нам нужны решения, которые могут преодолеть недостатки команды put и передать «потоковые данные» из генераторов данных в централизованные хранилища (особенно HDFS) с меньшей задержкой.

Проблема с HDFS

В HDFS файл существует как запись каталога, и длина файла будет считаться нулевой, пока он не будет закрыт. Например, если источник записывает данные в HDFS, и сеть была прервана в середине операции (без закрытия файла), данные, записанные в файле, будут потеряны.

Поэтому нам нужна надежная, настраиваемая и обслуживаемая система для передачи данных журнала в HDFS.

Примечание. В файловой системе POSIX всякий раз, когда мы обращаемся к файлу (скажем, выполняем операцию записи), другие программы все равно могут читать этот файл (по крайней мере, сохраненную часть файла). Это связано с тем, что файл существует на диске до его закрытия.

Доступные решения

Для отправки потоковых данных (файлов журналов, событий и т. Д.) Из различных источников в HDFS у нас есть следующие инструменты:

Писец в фейсбуке

Scribe — чрезвычайно популярный инструмент, который используется для агрегирования и потоковой передачи данных журнала. Он предназначен для масштабирования до очень большого количества узлов и может быть устойчивым к сбоям сети и узлов.

Апач Кафка

Кафка была разработана Apache Software Foundation. Это брокер сообщений с открытым исходным кодом. Используя Kafka, мы можем обрабатывать каналы с высокой пропускной способностью и низкой задержкой.

Apache Flume

Apache Flume — это механизм использования инструмента / службы / данных для сбора и передачи больших объемов потоковых данных, таких как данные журнала, события (и т. Д.), Из различных веб-серверов в централизованное хранилище данных.

Это высоконадежный, распределенный и настраиваемый инструмент, который в основном предназначен для передачи потоковых данных из различных источников в HDFS.

В этом уроке мы подробно обсудим, как использовать Flume с некоторыми примерами.

Apache Flume — Архитектура

Следующая иллюстрация изображает базовую архитектуру Flume. Как показано на рисунке, генераторы данных (такие как Facebook, Twitter) генерируют данные, которые собираются отдельными агентами Flume , работающими на них. После этого сборщик данных (который также является агентом) собирает данные от агентов, которые объединяются и помещаются в централизованное хранилище, такое как HDFS или HBase.

Flume Event

Событие является основной единицей данных, транспортируемых внутри Flume . Он содержит полезную нагрузку байтового массива, который должен быть перенесен из источника в пункт назначения с дополнительными заголовками. Типичное событие Flume будет иметь следующую структуру —

Flume Agent

Агент — это независимый демон-процесс (JVM) в Flume. Он получает данные (события) от клиентов или других агентов и перенаправляет их в следующий пункт назначения (приемник или агент). Flume может иметь более одного агента. Следующая диаграмма представляет Flume Agent

Как показано на схеме, Flume Agent содержит три основных компонента, а именно: источник , канал и приемник .

Источник

Источник — это компонент Агента, который получает данные от генераторов данных и передает их по одному или нескольким каналам в форме событий Flume.

Apache Flume поддерживает несколько типов источников, и каждый источник получает события от указанного генератора данных.

Пример — источник Avro, источник Thrift, источник Twitter 1% и т. Д.

канал

Канал — это временное хранилище, которое получает события от источника и буферизирует их до тех пор, пока они не будут поглощены приемниками. Он действует как мост между источниками и раковинами.

Эти каналы полностью транзакционны и могут работать с любым количеством источников и приемников.

Пример — канал JDBC, канал файловой системы, канал памяти и т. Д.

тонуть

Приемник хранит данные в централизованных хранилищах, таких как HBase и HDFS. Он потребляет данные (события) из каналов и доставляет их к месту назначения. Назначением приемника может быть другой агент или центральные магазины.

Пример — сток HDFS

Примечание . Агент распространения может иметь несколько источников, приемников и каналов. Мы перечислили все поддерживаемые источники, приемники, каналы в главе о конфигурации Flume этого руководства.

Дополнительные компоненты Flume Agent

То, что мы обсуждали выше, это примитивные компоненты агента. В дополнение к этому у нас есть еще несколько компонентов, которые играют жизненно важную роль в передаче событий из генератора данных в централизованные хранилища.

Перехватчики

Перехватчики используются для изменения / проверки событий потока, которые передаются между источником и каналом.

Селекторы каналов

Они используются для определения того, какой канал должен быть выбран для передачи данных в случае нескольких каналов. Есть два типа селекторов каналов —

• Селекторы каналов по умолчанию — Они также известны как копирующие селекторы каналов, они реплицируют все события в каждом канале.

• Мультиплексные селекторы каналов — они решают, что канал должен отправлять событие, основываясь на адресе в заголовке этого события.

Селекторы каналов по умолчанию — Они также известны как копирующие селекторы каналов, они реплицируют все события в каждом канале.

Мультиплексные селекторы каналов — они решают, что канал должен отправлять событие, основываясь на адресе в заголовке этого события.

Процессоры раковины

Они используются для вызова определенного приемника из выбранной группы приемников. Они используются для создания путей аварийного переключения для ваших приемников или событий балансировки нагрузки между несколькими приемниками из канала.

Apache Flume — Поток данных

Flume — это инфраструктура, которая используется для перемещения данных журнала в HDFS. Обычно события и данные журнала генерируются серверами журналов, и на этих серверах работают агенты Flume. Эти агенты получают данные от генераторов данных.

Данные в этих агентах будут собираться промежуточным узлом, известным как Collector . Как и у агентов, в Flume может быть несколько сборщиков.

Наконец, данные со всех этих сборщиков будут объединены и отправлены в централизованное хранилище, такое как HBase или HDFS. Следующая диаграмма объясняет поток данных в Flume.

Multi-hop Flow

Внутри Flume может быть несколько агентов, и до достижения конечного пункта назначения событие может проходить через более одного агента. Это известно как многопролетный поток .

Поток распада

Поток данных из одного источника в несколько каналов известен как поток разветвления . Это двух типов —

• Репликация — поток данных, в котором данные будут реплицироваться во всех настроенных каналах.

• Мультиплексирование — поток данных, в котором данные будут отправлены на выбранный канал, который упоминается в заголовке события.

Репликация — поток данных, в котором данные будут реплицироваться во всех настроенных каналах.

Мультиплексирование — поток данных, в котором данные будут отправлены на выбранный канал, который упоминается в заголовке события.

Поток вентилятора

Поток данных, в котором данные будут передаваться из многих источников в один канал, называется потоком данных .

Обработка ошибок

В Flume для каждого события выполняются две транзакции: одна у отправителя и одна у получателя. Отправитель отправляет события получателю. Вскоре после получения данных получатель совершает собственную транзакцию и отправляет «полученный» сигнал отправителю. После получения сигнала отправитель совершает свою транзакцию. (Отправитель не будет совершать транзакцию, пока не получит сигнал от получателя.)

Apache Flume — Окружающая среда

Мы уже обсуждали архитектуру Flume в предыдущей главе. В этой главе мы рассмотрим, как загрузить и настроить Apache Flume.

Прежде чем продолжить, вам необходимо иметь среду Java в вашей системе. Прежде всего, убедитесь, что в вашей системе установлена ​​Java. Для некоторых примеров в этом руководстве мы использовали Hadoop HDFS (в качестве приемника). Поэтому мы рекомендуем вам установить Hadoop вместе с Java. Для получения дополнительной информации перейдите по ссылке — http://www.tutorialspoint.com/hadoop/hadoop_enviornment_setup.htm.

Установка Flume

Прежде всего, загрузите последнюю версию программного обеспечения Apache Flume с веб-сайта https://flume.apache.org/ .

Шаг 1

Откройте сайт. Нажмите на ссылку для скачивания в левой части домашней страницы. Вы попадете на страницу загрузки Apache Flume.

Шаг 2

На странице загрузки вы можете увидеть ссылки на двоичные и исходные файлы Apache Flume. Нажмите на ссылку apache-flume-1.6.0-bin.tar.gz

Вы будете перенаправлены на список зеркал, где вы можете начать загрузку, щелкнув любое из этих зеркал. Таким же образом вы можете загрузить исходный код Apache Flume, нажав apache-flume-1.6.0-src.tar.gz .

Шаг 3

Создайте каталог с именем Flume в том же каталоге, где были установлены каталоги установки Hadoop , HBase и другого программного обеспечения (если вы его уже установили), как показано ниже.

$ mkdir Flume 

Шаг 4

Извлеките загруженные файлы tar, как показано ниже.

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

Шаг 5

Переместите содержимое файла apache- flume-1.6.0-bin.tar в созданный ранее каталог Flume, как показано ниже. (Предположим, мы создали каталог Flume для локального пользователя с именем Hadoop.)

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

Конфигурирование Flume

Чтобы настроить Flume, нам нужно изменить три файла: flume-env.sh, flumeconf.properties и bash.rc.

Установка пути / пути к классам

В файле .bashrc задайте домашнюю папку, путь и путь к классу для Flume, как показано ниже.

Папка конф

Если вы откроете папку conf Apache Flume, у вас будут следующие четыре файла:

• желоб-conf.properties.template,
• желоб-env.sh.template,
• flume-env.ps1.template и
• log4j.properties.

Теперь переименуйте

• файл flume -conf.properties.template как файл flume-conf.properties и

• flume-env.sh.template как flume-env.sh

файл flume -conf.properties.template как файл flume-conf.properties и

flume-env.sh.template как flume-env.sh

flume-env.sh

Откройте файл flume-env.sh и установите JAVA_Home в папку, где в вашей системе установлена ​​Java.

Проверка установки

Проверьте установку Apache Flume, просмотрев папку bin и введя следующую команду.

$ ./flume-ng 

Если вы успешно установили Flume, вы получите подсказку Flume, как показано ниже.

Apache Flume — Конфигурация

После установки Flume нам необходимо настроить его с помощью файла конфигурации, который является файлом свойств Java и содержит пары ключ-значение . Нам нужно передать значения ключам в файле.

В конфигурационном файле Flume нам нужно:

• Назовите компоненты текущего агента.
• Опишите / настройте источник.
• Опишите / настройте приемник.
• Опишите / настройте канал.
• Свяжите источник и приемник с каналом.

Обычно у нас может быть несколько агентов в Flume. Мы можем дифференцировать каждого агента, используя уникальное имя. И используя это имя, мы должны настроить каждого агента.

Наименование компонентов

Прежде всего, вам нужно назвать / перечислить компоненты, такие как источники, приемники и каналы агента, как показано ниже.

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

Flume поддерживает различные источники, приемники и каналы. Они перечислены в таблице ниже.

источники	каналы	Раковины
Avro Source Бережливый источник Exec Source Источник JMS Каталог каталогов Твиттер 1% пожарный шланг Источник Кафка Источник NetCat Source Источник Генератора Последовательности Syslog Sources Системный журнал TCP Source Многопортовый системный журнал TCP Source Системный журнал UDP Source HTTP Source Источник стресса Источники наследия Бережливый источник наследия Пользовательский источник Писец источник	Канал памяти Канал JDBC Канал Кафка Файловый канал Разлитый канал памяти Канал псевдо транзакций	Раковина HDFS Раковина улья Logger Sink Авро Раковина Береговая раковина IRC Раковина Рулонная мойка Нулевая Раковина HBaseSink AsyncHBaseSink MorphlineSolrSink ElasticSearchSink Kite Dataset Sink Раковина Кафки

Вы можете использовать любой из них. Например, если вы передаете данные Twitter с использованием источника Twitter через канал памяти в приемник HDFS и идентификатор агента в агенте TwitterAgent , то

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

После перечисления компонентов агента вы должны описать источник (и), приемник (и) и канал (ы), указав значения для их свойств.

Описание источника

Каждый источник будет иметь отдельный список свойств. Свойство с именем «type» является общим для каждого источника и используется для указания типа используемого нами источника.

Наряду со свойством «тип» необходимо указать значения всех необходимых свойств конкретного источника для его настройки, как показано ниже.

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

Например, если мы рассмотрим источник Twitter , ниже приведены свойства, которым мы должны предоставить значения для его настройки.

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

Описание раковины

Как и источник, каждый приемник будет иметь отдельный список свойств. Свойство с именем «тип» является общим для каждого приемника, и оно используется для указания типа используемого приемника. Наряду со свойством «тип» необходимо предоставить значения для всех необходимых свойств конкретного приемника, чтобы настроить его, как показано ниже.

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

Например, если мы рассмотрим приемник HDFS , ниже приведены свойства, которым мы должны предоставить значения для его настройки.

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

Описание канала

Flume предоставляет различные каналы для передачи данных между источниками и приемниками. Следовательно, наряду с источниками и каналами, необходимо описать канал, используемый в агенте.

Чтобы описать каждый канал, вам нужно установить необходимые свойства, как показано ниже.

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

Например, если мы рассмотрим канал памяти , ниже приведены свойства, которым мы должны предоставить значения для его настройки.

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

Привязка источника и раковины к каналу

Поскольку каналы соединяют источники и приемники, необходимо связать их обоих с каналом, как показано ниже.

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

В следующем примере показано, как связать источники и приемники с каналом. Здесь мы рассмотрим источник Twitter, канал памяти и приемник HDFS .

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

Запуск Flume Agent

После настройки мы должны запустить агент Flume. Это делается следующим образом —

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

где —

• agent — Команда для запуска Flume agent

• —conf, -c <conf> — использовать файл конфигурации в каталоге conf

• -f <файл> — указывает путь к файлу конфигурации, если он отсутствует

• —name, -n <name> — имя твиттера

• -D свойство = значение — устанавливает значение системного свойства Java.

agent — Команда для запуска Flume agent

—conf, -c <conf> — использовать файл конфигурации в каталоге conf

-f <файл> — указывает путь к файлу конфигурации, если он отсутствует

—name, -n <name> — имя твиттера

-D свойство = значение — устанавливает значение системного свойства Java.

Apache Flume — получение данных из Twitter

Используя Flume, мы можем получать данные из различных служб и транспортировать их в централизованные хранилища (HDFS и HBase). В этой главе объясняется, как получать данные из службы Twitter и сохранять их в HDFS с помощью Apache Flume.

Как обсуждалось в Flume Architecture, веб-сервер генерирует данные журнала, и эти данные собираются агентом в Flume. Канал буферизует эти данные в приемник, который в конечном итоге передает их в централизованные хранилища.

В примере, приведенном в этой главе, мы создадим приложение и получим от него твиты, используя экспериментальный источник Twitter, предоставленный Apache Flume. Мы будем использовать канал памяти для буферизации этих твитов, а приемник HDFS для отправки этих твитов в HDFS.

Чтобы получить данные из Twitter, нам нужно будет выполнить следующие шаги:

• Создать приложение для Twitter
• Установить / запустить HDFS
• Настроить Flume

Создание приложения Twitter

Чтобы получать твиты из Twitter, нужно создать приложение для Twitter. Следуйте инструкциям ниже, чтобы создать приложение для Twitter.

Шаг 1

Чтобы создать приложение для Twitter, нажмите следующую ссылку https://apps.twitter.com/ . Войдите в свой аккаунт Twitter. У вас будет окно управления приложениями Twitter, где вы можете создавать, удалять и управлять приложениями Twitter.

Шаг 2

Нажмите на кнопку « Создать новое приложение» . Вы будете перенаправлены в окно, где вы получите форму заявки, в которой вы должны будете заполнить свои данные, чтобы создать приложение. При заполнении адреса сайта укажите полный шаблон URL, например, http://example.com.

Шаг 3

Заполните детали, примите Соглашение с разработчиком, когда закончите, нажмите кнопку « Создать приложение в Твиттере», которая находится внизу страницы. Если все идет хорошо, приложение будет создано с указанными деталями, как показано ниже.

Шаг 4

Под ключами и вкладкой токены доступа внизу страницы вы можете увидеть кнопку под названием Создать мой токен доступа . Нажмите на него, чтобы сгенерировать токен доступа.

Шаг 5

Наконец, нажмите на кнопку Test OAuth , которая находится в правой верхней части страницы. Это приведет к появлению страницы с вашим ключом потребителя, секретом потребителя, токеном доступа и секретом токена доступа . Скопируйте эти детали. Они полезны для настройки агента в Flume.

Запуск HDFS

Поскольку мы храним данные в HDFS, нам нужно установить / проверить Hadoop. Запустите Hadoop и создайте в нем папку для хранения данных Flume. Выполните шаги, приведенные ниже, перед настройкой Flume.

Шаг 1. Установите / проверьте Hadoop

Установите Hadoop . Если Hadoop уже установлен в вашей системе, проверьте установку с помощью команды версии Hadoop, как показано ниже.

$ hadoop version 

Если ваша система содержит Hadoop и вы установили переменную path, вы получите следующий вывод:

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

Шаг 2: Запуск Hadoop

Просмотрите каталог sbin Hadoop и запустите yarn и Hadoop dfs (распределенную файловую систему), как показано ниже.

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 

Шаг 3: Создайте каталог в HDFS

В Hadoop DFS вы можете создавать каталоги с помощью команды mkdir . Просмотрите его и создайте каталог с именем twitter_data в нужном пути, как показано ниже.

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

Конфигурирование Flume

Мы должны настроить источник, канал и приемник, используя файл конфигурации в папке conf . В примере, приведенном в этой главе, используется экспериментальный источник, предоставленный Apache Flume, под названием « Twitter 1% Firehose Memory channel» и приемник HDFS.

Твиттер 1% Firehose Source

Этот источник очень экспериментален. Он подключается к 1% -ному примеру Twitter Firehose с использованием потокового API и непрерывно загружает твиты, преобразует их в формат Avro и отправляет события Avro в нисходящий приемник Flume.

Мы получим этот источник по умолчанию вместе с установкой Flume. Файлы jar, соответствующие этому источнику, могут быть расположены в папке lib, как показано ниже.

Установка пути к классам

Установите переменную classpath для папки lib Flume в файле Flume-env.sh, как показано ниже.

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

Для этого источника требуются такие сведения, как ключ потребителя, секрет потребителя, токен доступа и секрет токена доступа приложения Twitter. При настройке этого источника вы должны указать следующие свойства:

• каналы

• Тип источника: org.apache.flume.source.twitter.TwitterSource

• consumerKey — ключ пользователя OAuth

• consumerSecret — OAuth, секрет пользователя

• accessToken — токен доступа OAuth

• accessTokenSecret — Секретный токен OAuth

• maxBatchSize — максимальное количество сообщений в Твиттере, которое должно быть в пакете твиттера. Значение по умолчанию 1000 (необязательно).

• maxBatchDurationMillis — Максимальное количество миллисекунд ожидания до закрытия пакета. Значение по умолчанию 1000 (необязательно).

каналы

Тип источника: org.apache.flume.source.twitter.TwitterSource

consumerKey — ключ пользователя OAuth

consumerSecret — OAuth, секрет пользователя

accessToken — токен доступа OAuth

accessTokenSecret — Секретный токен OAuth

maxBatchSize — максимальное количество сообщений в Твиттере, которое должно быть в пакете твиттера. Значение по умолчанию 1000 (необязательно).

maxBatchDurationMillis — Максимальное количество миллисекунд ожидания до закрытия пакета. Значение по умолчанию 1000 (необязательно).

канал

Мы используем канал памяти. Чтобы настроить канал памяти, необходимо указать значение для типа канала.

• тип — содержит тип канала. В нашем примере типом является MemChannel .

• Емкость — это максимальное количество событий, хранящихся в канале. Его значение по умолчанию — 100 (необязательно).

• TransactionCapacity — это максимальное количество событий, которые канал принимает или отправляет. Его значение по умолчанию — 100 (необязательно).

тип — содержит тип канала. В нашем примере типом является MemChannel .

Емкость — это максимальное количество событий, хранящихся в канале. Его значение по умолчанию — 100 (необязательно).

TransactionCapacity — это максимальное количество событий, которые канал принимает или отправляет. Его значение по умолчанию — 100 (необязательно).

Раковина HDFS

Этот приемник записывает данные в HDFS. Чтобы настроить этот приемник, вы должны предоставить следующую информацию.

• канал

• тип — hdfs

• hdfs.path — путь к каталогу в HDFS, в котором должны храниться данные.

канал

тип — hdfs

hdfs.path — путь к каталогу в HDFS, в котором должны храниться данные.

И мы можем предоставить некоторые дополнительные значения в зависимости от сценария. Ниже приведены необязательные свойства приемника HDFS, которые мы настраиваем в нашем приложении.

• fileType — это необходимый формат файла нашего HDFS. SequenceFile, DataStream и CompressedStream являются тремя типами, доступными для этого потока. В нашем примере мы используем DataStream .

• writeFormat — может быть либо текстовым, либо записываемым.

• batchSize — это число событий, записанных в файл перед его сбросом в HDFS. Его значение по умолчанию равно 100.

• rollsize — размер файла для запуска броска. Это значение по умолчанию составляет 100.

• rollCount — количество событий, записанных в файл перед его прокруткой. Значение по умолчанию 10.

fileType — это необходимый формат файла нашего HDFS. SequenceFile, DataStream и CompressedStream являются тремя типами, доступными для этого потока. В нашем примере мы используем DataStream .

writeFormat — может быть либо текстовым, либо записываемым.

batchSize — это число событий, записанных в файл перед его сбросом в HDFS. Его значение по умолчанию равно 100.

rollsize — размер файла для запуска броска. Это значение по умолчанию составляет 100.

rollCount — количество событий, записанных в файл перед его прокруткой. Значение по умолчанию 10.

Пример — файл конфигурации

Ниже приведен пример файла конфигурации. Скопируйте этот контент и сохраните как twitter.conf в папке conf 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 

выполнение

Просмотрите домашний каталог Flume и запустите приложение, как показано ниже.

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

Если все пойдет хорошо, начнется потоковая передача твитов в HDFS. Ниже приведен снимок окна командной строки при получении твитов.

Проверка HDFS

Вы можете получить доступ к веб-интерфейсу администрирования Hadoop, используя приведенный ниже URL-адрес.

http://localhost:50070/ 

Нажмите на раскрывающееся меню « Утилиты» в правой части страницы. Вы можете увидеть два варианта, как показано на снимке экрана, приведенном ниже.

Нажмите Обзор файловой системы и введите путь к каталогу HDFS, в котором вы сохранили твиты. В нашем примере путь будет / user / Hadoop / twitter_data / . Затем вы можете увидеть список файлов журнала Twitter, хранящихся в HDFS, как показано ниже.

Apache Flume — источник генератора последовательностей

В предыдущей главе мы увидели, как извлекать данные из твиттерного источника в HDFS. Эта глава объясняет, как получить данные из генератора последовательностей .

Предпосылки

Для запуска примера, представленного в этой главе, вам необходимо установить HDFS вместе с Flume . Поэтому проверьте установку Hadoop и запустите HDFS, прежде чем продолжить. (Обратитесь к предыдущей главе, чтобы узнать, как запустить HDFS).

Конфигурирование Flume

Мы должны настроить источник, канал и приемник, используя файл конфигурации в папке conf . Пример, приведенный в этой главе, использует источник генератора последовательности , канал памяти и приемник HDFS .

Источник Генератора Последовательности

Это источник, который генерирует события непрерывно. Он поддерживает счетчик, который начинается с 0 и увеличивается на 1. Он используется для целей тестирования. При настройке этого источника вы должны указать следующие свойства:

• каналы

• Тип источника — seq

каналы

Тип источника — seq

канал

Мы используем канал памяти . Чтобы настроить канал памяти, необходимо указать значение для типа канала. Ниже приведен список свойств, которые необходимо указать при настройке канала памяти.

• тип — содержит тип канала. В нашем примере типом является MemChannel.

• Емкость — это максимальное количество событий, хранящихся в канале. Его значение по умолчанию — 100. (необязательно)

• TransactionCapacity — это максимальное количество событий, которые канал принимает или отправляет. По умолчанию это 100. (необязательно).

тип — содержит тип канала. В нашем примере типом является MemChannel.

Емкость — это максимальное количество событий, хранящихся в канале. Его значение по умолчанию — 100. (необязательно)

TransactionCapacity — это максимальное количество событий, которые канал принимает или отправляет. По умолчанию это 100. (необязательно).

Раковина HDFS

Этот приемник записывает данные в HDFS. Чтобы настроить этот приемник, вы должны предоставить следующую информацию.

• канал

• тип — hdfs

• hdfs.path — путь к каталогу в HDFS, в котором должны храниться данные.

канал

тип — hdfs

hdfs.path — путь к каталогу в HDFS, в котором должны храниться данные.

И мы можем предоставить некоторые дополнительные значения в зависимости от сценария. Ниже приведены необязательные свойства приемника HDFS, которые мы настраиваем в нашем приложении.

• fileType — это необходимый формат файла нашего HDFS. SequenceFile, DataStream и CompressedStream являются тремя типами, доступными для этого потока. В нашем примере мы используем DataStream .

• writeFormat — может быть либо текстовым, либо записываемым.

• batchSize — это число событий, записанных в файл перед его сбросом в HDFS. Его значение по умолчанию равно 100.

• rollsize — размер файла для запуска броска. Это значение по умолчанию составляет 100.

• rollCount — количество событий, записанных в файл перед его прокруткой. Значение по умолчанию 10.

fileType — это необходимый формат файла нашего HDFS. SequenceFile, DataStream и CompressedStream являются тремя типами, доступными для этого потока. В нашем примере мы используем DataStream .

writeFormat — может быть либо текстовым, либо записываемым.

batchSize — это число событий, записанных в файл перед его сбросом в HDFS. Его значение по умолчанию равно 100.

rollsize — размер файла для запуска броска. Это значение по умолчанию составляет 100.

rollCount — количество событий, записанных в файл перед его прокруткой. Значение по умолчанию 10.

Пример — файл конфигурации

Ниже приведен пример файла конфигурации. Скопируйте этот контент и сохраните как seq_gen .conf в папке conf 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 

выполнение

Просмотрите домашний каталог Flume и запустите приложение, как показано ниже.

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

Если все идет хорошо, источник начинает генерировать порядковые номера, которые будут помещены в HDFS в виде файлов журнала.

Ниже приведен снимок окна командной строки, извлекающего данные, сгенерированные генератором последовательностей, в HDFS.

Проверка HDFS

Вы можете получить доступ к веб-интерфейсу администрирования Hadoop, используя следующий URL-адрес:

http://localhost:50070/

Нажмите на раскрывающееся меню « Утилиты» в правой части страницы. Вы можете увидеть два варианта, как показано на диаграмме, приведенной ниже.

Нажмите Обзор файловой системы и введите путь к каталогу HDFS, в котором вы сохранили данные, сгенерированные генератором последовательности.

В нашем примере путь будет / user / Hadoop / seqgen_data / . Затем вы можете увидеть список файлов журнала, сгенерированных генератором последовательности, которые хранятся в HDFS, как указано ниже.

Проверка содержимого файла

Все эти файлы журнала содержат числа в последовательном формате. Вы можете проверить содержимое этого файла в файловой системе, используя команду cat, как показано ниже.

Apache Flume — NetCat Source

В этой главе приведен пример, объясняющий, как вы можете генерировать события и впоследствии регистрировать их в консоли. Для этого мы используем источник NetCat и приемник логгера .

Предпосылки

Для запуска примера, приведенного в этой главе, вам необходимо установить Flume .

Конфигурирование Flume

Мы должны настроить источник, канал и приемник, используя файл конфигурации в папке conf . Пример, приведенный в этой главе, использует источник NetCat, канал памяти и приемник логгера .

NetCat Source

При настройке источника NetCat мы должны указать порт при настройке источника. Теперь источник (источник NetCat) прослушивает данный порт и получает каждую строку, которую мы ввели в этот порт, как отдельное событие и передает его в приемник по указанному каналу.

При настройке этого источника вы должны указать следующие свойства:

• каналы

• Тип источника — netcat

• bind — имя хоста или IP-адрес для привязки.

• порт — номер порта, к которому мы хотим, чтобы источник прослушивал.

каналы

Тип источника — netcat

bind — имя хоста или IP-адрес для привязки.

порт — номер порта, к которому мы хотим, чтобы источник прослушивал.

канал

Мы используем канал памяти . Чтобы настроить канал памяти, необходимо указать значение для типа канала. Ниже приведен список свойств, которые необходимо указать при настройке канала памяти.

• тип — содержит тип канала. В нашем примере типом является MemChannel .

• Емкость — это максимальное количество событий, хранящихся в канале. Его значение по умолчанию — 100. (необязательно)

• TransactionCapacity — это максимальное количество событий, которые канал принимает или отправляет. Его значение по умолчанию — 100. (необязательно).

тип — содержит тип канала. В нашем примере типом является MemChannel .

Емкость — это максимальное количество событий, хранящихся в канале. Его значение по умолчанию — 100. (необязательно)

TransactionCapacity — это максимальное количество событий, которые канал принимает или отправляет. Его значение по умолчанию — 100. (необязательно).

Logger Sink

Этот приемник регистрирует все события, переданные ему. Как правило, он используется для целей тестирования или отладки. Чтобы настроить этот приемник, вы должны предоставить следующую информацию.

• канал

• тип — регистратор

канал

тип — регистратор

Пример файла конфигурации

Ниже приведен пример файла конфигурации. Скопируйте этот контент и сохраните как netcat.conf в папке conf 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

выполнение

Просмотрите домашний каталог Flume и запустите приложение, как показано ниже.

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

Если все идет хорошо, источник начинает прослушивать данный порт. В данном случае это 56565 . Ниже приведен снимок окна командной строки источника NetCat, который запустился и прослушивает порт 56565.

Передача данных в источник

Чтобы передать данные в источник NetCat, вы должны открыть порт, указанный в файле конфигурации. Откройте отдельный терминал и подключитесь к источнику (56565) с помощью команды curl . После успешного подключения вы получите сообщение « подключено », как показано ниже.

$ curl telnet://localhost:56565 
connected 

Теперь вы можете вводить свои данные построчно (после каждой строки вы должны нажать Enter). Источник NetCat получает каждую строку как отдельное событие, и вы получите полученное сообщение « ОК ».

Всякий раз, когда вы закончите с передачей данных, вы можете выйти из консоли, нажав ( Ctrl & plus; C ). Ниже приведен снимок консоли, где мы подключились к источнику с помощью команды curl .

Каждая строка, введенная в вышеприведенной консоли, будет получена источником как отдельное событие. Поскольку мы использовали приемник Logger , эти события будут регистрироваться на консоли (исходной консоли) через указанный канал (в данном случае канал памяти).

На следующем снимке показана консоль NetCat, в которой регистрируются события.