1. Что такое Flume?

2. Основные термины

• Событие (event) — единица данных с дополнительной мета-информацией. По структуре событие напоминает POST-запрос.
  
  • Заголовки (headers) — мета-информация, набор пар “ключ”-”значение”.
  • Содержимое (body) — собственно, данные, ради передачи которых всё затевается. Передается как byte[].
  
  
• Клиент (client) — внешний, по отношению к узлу Flume, сервис, поставляющий данные.
  
  
• Источник (source) — отвечает за прием данных. При этом Flume предусматривает два типа источников — EventDrivenSource и PollableSource. В первом случае источник сам решает, когда добавлять события в канал (например, HTTPSource добавляет события по мере получения HTTP-запросов). PollableSource в противовес EventDrivenSource является пассивным — Flume просто периодически опрашивает источник на предмет появления новых событий.
  
  
• Сток (Sink) — компонент, отвечающий за передачу данных на следующий этап обработки. Это может быть другой узел Flume, файловая система, HDFS и т.п.
  
  
• Канал (channel) — компонент, выполняющий роль буфера при транспортировке данных. Канал является пассивным компонентом, он самостоятельно не инициирует никаких действий. Источники добавляют события в канал, в то время как стоки его опустошают.
  
  
• Агент (agent) — процесс, в рамках которого функционируют компоненты Flume (источники, каналы, стоки). JVM Instance, в общем. Один узел может содержать несколько агентов.

3. Структура узла Flume

• Узел может иметь в своем составе множество источников, каналов и стоков.
• Один источник может складывать события в несколько каналов (дублировать или распределять по некоторому правилу).
• Несколько источников складывать события в один канал.
• Один сток может работать только с одним каналом.
• Несколько стоков могут забирать события из одного канала (равномерно или по некоторому правилу балансировки).

4. Конфигурация и запуск узла Flume

• Memchannel (канал, использующий оперативную память для хранения событий).
• NetCat Source.
• Logger Sink (сток, выводящий события в консоль).

### ==================== Компоненты узла ==================== ###
# Перечисляем все основные компоненты, из которых будет состоять наш узел: источники, каналы и узлы

# <agent>.sources - имена источников, разделенные пробелом (в этом примере один источник: my_source)
my-agent.sources = my-source
# <agent>.channels - аналогично указываем имена каналов
my-agent.channels = my-channel
# <agent>.sinks - для стоков то же самое
my-agent.sinks = my-sink

### ==================== Источник my_source ================== ###
# Тип источника - netcat (источники из стандартной поставки Flume имеют зарезервированные имена-псевдонимы,
# в общем случае здесь можно указать полное имя класса источника, в т.ч., вашего собственного)
my-agent.sources.my-source.type = netcat
# Указываем, куда биндить наш исчтоник
my-agent.sources.my-source.bind = 0.0.0.0
my-agent.sources.my-source.port = 11111
# Указываем источнику канал (или список каналов, через пробел), куда отправлять полученные события
my-agent.sources.my-source.channels = my-channel

### ==================== Канал my_channel ================== ###
# Используем тип канала из пакета Flume - memory (как и с источником, здесь можно казать свой класс), который хранит события в памяти
my-agent.channels.my-channel.type = memory
# Вместимость канала, кол_во событий
my-agent.channels.my-channel.capacity = 10000
# Число событий в одной транзакции (как на добавление, так и на "вытягивание")
my-agent.channels.my-channel.transactionCapacity = 100

### ==================== Сток my_sink ================== ###
# Тип стока - логгер, пишуший события в консоль (и здесь также можно указать свой класс)
my-agent.sinks.my-sink.type = logger
# Из какого канала будем забирать события
my-agent.sinks.my-sink.channel = my-channel
# Настройка исключительно для стока типа logger - сколько первых байт тела события выводить в консоль
my-agent.sinks.my-sink.maxBytesToLog = 256

1. На кластере Hadoop, через Cloudera Manager (в этой статье есть подробное описание того, как это сделать).
2. Как Java-сервис, используя библиотеки Flume.

    <repositories>     
        <repository>
            <id>cloudera</id>
            <url>https://repository.cloudera.com/artifactory/cloudera-repos/</url>
        </repository>
    </repositories>
 
    <dependencies>        
        <dependency>
            <groupId>org.apache.flume</groupId>
            <artifactId>flume-ng-sdk</artifactId>
            <version>1.5.0-cdh5.3.0</version>
        </dependency>
        <dependency>
            <groupId>org.apache.flume</groupId>
            <artifactId>flume-ng-node</artifactId>
            <version>1.5.0-cdh5.3.0</version>
        </dependency>
    </dependencies>

package ru.flume.samples;

import org.apache.flume.node.Application;

public class FlumeLauncher {
    
    public static void main(String[] args) {  
        // файл с конфигурацией Log4j я позволю себе указать прямо здесь
        System.setProperty("log4j.configuration", "file:/flume/config/log4j.properties");          
        
        // Запускаем Flume с параметрами:
        Application.main(new String[]{
            "-f", "/flume/config/sample.conf", // путь до файла с конфигурацией
            "-n", "my-agent"                   // имя агента
        });   
    }    
}

INFO  main conf.FlumeConfiguration - Processing:my-sink
INFO  main conf.FlumeConfiguration - Added sinks: my-sink Agent: my-agent
INFO  main conf.FlumeConfiguration - Processing:my-sink
INFO  main conf.FlumeConfiguration - Processing:my-sink
INFO  main conf.FlumeConfiguration - Post-validation flume configuration contains configuration for agents: [my-agent]
INFO  main node.AbstractConfigurationProvider - Creating channels
INFO  main channel.DefaultChannelFactory - Creating instance of channel my-channel type memory
INFO  main node.AbstractConfigurationProvider - Created channel my-channel
INFO  main source.DefaultSourceFactory - Creating instance of source my-source, type netcat
INFO  main sink.DefaultSinkFactory - Creating instance of sink: my-sink, type: logger
INFO  main node.AbstractConfigurationProvider - Channel my-channel connected to [my-source, my-sink]
INFO  main node.Application - Starting new configuration:
{
    sourceRunners:{
        my-source=EventDrivenSourceRunner: { 
            source:org.apache.flume.source.NetcatSource{
                name:my-source,
                state:IDLE
            }
        }
    } 
    sinkRunners:{
        my-sink=SinkRunner: { 
            policy:org.apache.flume.sink.DefaultSinkProcessor@77f03bb1 counterGroup:{ name:null counters:{} } 
        }
    }            
    channels:{
        my-channel=org.apache.flume.channel.MemoryChannel{
            name: my-channel
        }
    }
}
INFO  main node.Application - Starting Channel my-channel
INFO  main node.Application - Waiting for channel: my-channel to start. Sleeping for 500 ms
INFO  lifecycleSupervisor-1-0 instrumentation.MonitoredCounterGroup - Monitored counter group for type: CHANNEL, name: my-channel: Successfully registered new MBean.
INFO  lifecycleSupervisor-1-0 instrumentation.MonitoredCounterGroup - Component type: CHANNEL, name: my-channel started
INFO  main node.Application - Starting Sink my-sink
INFO  main node.Application - Starting Source my-source
INFO  lifecycleSupervisor-1-1 source.NetcatSource - Source starting
INFO  lifecycleSupervisor-1-1 source.NetcatSource - Created serverSocket:sun.nio.ch.ServerSocketChannelImpl[/0:0:0:0:0:0:0:0:11111]

Message 1
Message 2
Message 3

nc 127.0.0.1 11111 < test.txt

sink.LoggerSink - Event: { headers:{} body: 4D 65 73 73 61 67 65 20 31 0D                   Message 1. }
sink.LoggerSink - Event: { headers:{} body: 4D 65 73 73 61 67 65 20 32 0D                   Message 2. }
sink.LoggerSink - Event: { headers:{} body: 4D 65 73 73 61 67 65 20 33 0D                   Message 3. }

5. Цепочка узлов Flume

node1.sources = my-source
node1.channels = my-channel
# Теперь здесь 2 стока:
node1.sinks = my-sink1 my-sink2

node1.sources.my-source.type = netcat
node1.sources.my-source.bind = 0.0.0.0
node1.sources.my-source.port = 11111
node1.sources.my-source.channels  = my-channel

node1.channels.my-channel.type = memory
node1.channels.my-channel.capacity = 10000
node1.channels.my-channel.transactionCapacity = 100

# Оба стока делаем с типом avro, они будут опустошать наш единственный канал вдвоем
# Хосты принимающих узлов я оставлю локальными, чтобы всю эту цепочку можно было
# попробовать запустить на одной машине
node1.sinks.my-sink1.type = avro
node1.sinks.my-sink1.channel = my-channel
node1.sinks.my-sink1.hostname = 127.0.0.1
node1.sinks.my-sink1.port = 11112
node1.sinks.my-sink1.batch-size = 100

node1.sinks.my-sink2.type = avro
node1.sinks.my-sink2.channel = my-channel
node1.sinks.my-sink2.hostname = 127.0.0.1
node1.sinks.my-sink2.port = 11113
node1.sinks.my-sink2.batch-size = 100

node2.sources = my-source
node2.channels = my-channel
node2.sinks = my-sink

# Поскольку на узле 1 сток имеет тип avro, здесь мы указываем источник типа avro
node2.sources.my-source.type = avro
node2.sources.my-source.bind = 0.0.0.0
node2.sources.my-source.port = 11112
node2.sources.my-source.channels  = my-channel

node2.channels.my-channel.type = memory
node2.channels.my-channel.capacity = 10000
node2.channels.my-channel.transactionCapacity = 100

node2.sinks.my-sink.type = logger
node2.sinks.my-sink.channel = my-channel
node2.sinks.my-sink.maxBytesToLog = 256

node3.sources = my-source
node3.channels = my-channel
node3.sinks = my-sink

# Поскольку на узле 1 сток имеет тип avro, здесь мы указываем источник типа avro
node3.sources.my-source.type = avro
node3.sources.my-source.bind = 0.0.0.0
node3.sources.my-source.port = 11113
node3.sources.my-source.channels  = my-channel

node3.channels.my-channel.type = memory
node3.channels.my-channel.capacity = 10000
node3.channels.my-channel.transactionCapacity = 100

node3.sinks.my-sink.type = logger
node3.sinks.my-sink.channel = my-channel
node3.sinks.my-sink.maxBytesToLog = 256

Application.main(new String[]{"-f", "/flume/config/node1.conf", "-n", "node1"});  
    // для других узлов по аналогии:
    //Application.main(new String[]{"-f", "/flume/config/node2.conf", "-n", "node2"});  
    //Application.main(new String[]{"-f", "/flume/config/node3.conf", "-n", "node3"});

Заключение

• Использование заголовков и канальных селекторов (Channel Selector).
• «Боевые» компоненты Flume:
  • Avro Source;
  • File Channel;
  • Avro Sink;
  • HDFS Sink;
  • File Roll Sink.
• Мониторинг состояния узла Flume.

Использованные источники и полезные ссылки

• Официальная страница Apache Flume
• Официальный гайд по настройке компонентов Flume
• Hadoop, часть 2: сбор данных через Flume | Блог компании Селектел — статья о настройке Flume средствами Cloudera.
• Hari Shreedharan: Using Flume — неплохая книга, описывающая возможности Flume.