Flume: HDFSEventSink - как мультиплексировать динамически?

Question

Резюме: У меня есть сценарий мультиплексирования, и вы хотели бы знать, как мультиплексировать динамически - не на основе статически настроенного значения, а на основе значения переменной поля (например, даты).

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

agent.sources.jmsSource.channels = chan-10 chan-11 # ... 

agent.sources.jmsSource.selector.type = multiplexing 
agent.sources.jmsSource.selector.header = EntityId 
agent.sources.jmsSource.selector.mapping.10 = chan-10 
agent.sources.jmsSource.selector.mapping.11 = chan-11 
# ... 

Каждый из каналов идет в отдельном HDFSEventSink, "hdfsSink-N":

agent.sinks.hdfsSink-10.channel = chan-10 
agent.sinks.hdfsSink-10.hdfs.path = hdfs://some/path/ 
agent.sinks.hdfsSink-10.hdfs.filePrefix = entity10 
# ... 
agent.sinks.hdfsSink-11.channel = chan-11 
agent.sinks.hdfsSink-11.hdfs.path = hdfs://some/path/ 
agent.sinks.hdfsSink-11.hdfs.filePrefix = entity11 
# ... 

Это создает файл для каждого объекта, который хорошо. Теперь я хочу, чтобы ввести вторую переменную, которая является динамической: дата. В зависимости от даты события, я хочу, чтобы создавать файлы на сущность на дату. Дата - это динамическое значение, поэтому я не могу предварительно сконфигурировать несколько приемников, поэтому каждый отправляет в отдельный файл. Кроме того, вы можете указать только один HDFS-выход для Sink.

Итак, это было похоже на «множественные выходы HDFSEventSink» (аналогично библиотеке MultipleOutputs от Hadoop). Есть ли такая функциональность в Flume?

Если нет, есть ли какой-либо изящный способ исправить это или работать над этим? Другой вариант - изменить HDFSEventSink, и, похоже, его можно реализовать, создав другое создание «realName» (String) для каждого события.

Answer 1

Фактически вы можете указать переменную в вашем пути hdfs sink или filePrefix. Например, если ключ переменной является «датой» в заголовках событий, вы можете настроить следующим образом:

agent.sinks.hdfsSink-11.hdfs.filePrefix = entity11-%{date}