Map or Reduce?

[Palych]

Рисовал я намедни читалку логов.
Использовал Java Streams, вызывал их из Javascript/Nashorn, но это не важно...
Задачка наверное весьма типическая, я просто чайник в этой области.

Логи построены просто: сначала служебные поля (timestamp, location, etc.), потом собственно сообщение.
Идея простая: пропускаем это через mapper, который парсит эти строчки в структуры с полями timestamp, location, ..., message.
А потом уже фильруем, редьюсим...

Засада в том, что сообщение может быть в несколько строк. То есть - количество строк на входе может быть больше чем записей на выходе.
Я сделал просто, замёл под ковёр: сделал mapper Statefull (оскоромился...), т.е. складывал неоконченные записи в field variable и возвращал null пока не дочитаю запись. А сразу да маппером поставил фильтр, который не пропускал нули.

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

[iDesperado]

на классический мап-редюс мне кажется такое не кладется, т.к. у тебя какие-то взаимосвязи между строками возникают (пол сообщения тут пол там). идея мап-редюс в том что может порубить входной файл как звезды укажут и раскидать на разные мапперы, а в твоем варианте пол сообщения на одном мапере окажется, половина без части критических данных на другом.
наверно однопоточный "mapper Statefull" через попу, но вариант в такой ситуации.

[Сабина]

интересный лог.. а почему сам record в несколько строк ? мейнфрейм что ли ?

[Ann4Ann]

я когда-то делала Logstash -> Elasticsearch, а там уже кастом агрегаторы шуршали. Но задача была не просто лог собрать, а навести там туману всякого, типа feedback loop

[Мальчик-Одуванчик]

интересный лог.. а почему сам record в несколько строк ? мейнфрейм что ли ?

Наверное чтобы его читать можно было.

[Kolbasoff]

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

Т.е. есть группа строчек, помеченные одним тэгом? Тогда так:

read()
.map(line => (tag, line))
.aggregateByKey(Initializer)(Mapper: (String => StringBuffer), Reducer:( StringBuffer, StringBuffer) => StringBuffer)
.map( (tag, StringBuffer) => (String))
.write();

Initializer создает буфер для строки собранной из разных строк с одним тэгом
Mapper помещает строчку в этот буфер
Reducer смотрит на два буфера и, в зависимости от их контента, соединяет их правильным образом.
Логика соединения двух буферов должна быть независимой от порядка их поступления в Reducer (коммутативна и ассоциативна, да!).
Количество строчек под одним tag может быть любым. Чем больше строчек тем затейливей логика в Reducer.

финальный map очищает tuple2 (tag, StringBuffer) от tag и выделяет содержимое из StringBuffer.