Я тестирую flume для загрузки данных в hHase и думаю о параллельной загрузке данных с использованием селектора и inteceptor flume из-за разницы в скорости между источником и приемником.

Итак, что я хочу сделать с Flume

создание заголовка события с типом перехватчиков regex_extractor

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

в одном источнике-канале-приемнике.

и попробовал конфигурацию как ниже.

    agent.sources = tailsrc
    agent.channels = mem1 mem2
    agent.sinks = std1 std2
    agent.sources.tailsrc.type = exec
    agent.sources.tailsrc.command = tail -F /home/flumeuser/test/in.txt
    agent.sources.tailsrc.batchSize = 1

    agent.sources.tailsrc.interceptors = i1
    agent.sources.tailsrc.interceptors.i1.type = regex_extractor
    agent.sources.tailsrc.interceptors.i1.regex = ^(\\d)
    agent.sources.tailsrc.interceptors.i1.serializers = t1
    agent.sources.tailsrc.interceptors.i1.serializers.t1.name = type

    agent.sources.tailsrc.selector.type = multiplexing
    agent.sources.tailsrc.selector.header = type
    agent.sources.tailsrc.selector.mapping.1 = mem1
    agent.sources.tailsrc.selector.mapping.2 = mem2

    agent.sinks.std1.type = file_roll
    agent.sinks.std1.channel = mem1
    agent.sinks.std1.batchSize = 1
    agent.sinks.std1.sink.directory = /var/log/flumeout/1
    agent.sinks.std1.rollInterval = 0

    agent.sinks.std2.type = file_roll
    agent.sinks.std2.channel = mem2
    agent.sinks.std2.batchSize = 1
    agent.sinks.std2.sink.directory = /var/log/flumeout/2
    agent.sinks.std2.rollInterval = 0

    agent.channels.mem1.type = memory
    agent.channels.mem1.capacity = 100

    agent.channels.mem2.type = memory
    agent.channels.mem2.capacity = 100

Но это не работает!

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

Есть ли какое-то неправильное выражение или что-то, что я пропустил?

Спасибо за чтение. :)