Hazelcast Jet отбрасывает пустые результаты агрегации - PullRequest
Новая
       25

Hazelcast Jet отбрасывает пустые результаты агрегации

0 голосов
/ 25 апреля 2018

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

        Pipeline pipeline = Pipeline.create();
        pipeline.drawFrom(Sources.<Long, Foo>map("map"))
                .map(Map.Entry::getValue)
                .addTimestamps(Foo::getTimeMillisecond, LIMIT)
                .window(WindowDefinition.sliding(100, 10))
                .aggregate(FooAggregateOperations.aggregateFoo(), (s, e, r) -> {
                    return String.format("started: %s\n%s\nended: %s\n", s, r, e);
                })
                .drainTo(Sinks.files(sinkDirectory));

Как вы видите, агрегатор возвращает строку: 

public class FooAggregateOperations {

    public static AggregateOperation1<Foo, FooAccumulator, String> aggregateFoo() {
        return AggregateOperation
                .withCreate(FooAccumulator::new)
                .andAccumulate(FooAggregateOperations::accumulate)
                .andCombine(FooAggregateOperations::combine)
                .andDeduct(FooAggregateOperations::deduct)
                .andFinish(FooAccumulator::getResult);
    }
}

Вопрос в основном, как можно отбросить игнорируемые окна / результаты агрегации, прежде чем они будут объединены / вычтены с другими результатами или сброшены в сток?

1 Ответ

0 голосов
/ 25 апреля 2018

Чтобы отфильтровать пустые результаты агрегации, вы можете использовать следующий подход: 

    Pipeline pipeline = Pipeline.create();
    pipeline.drawFrom(Sources.<Long, Foo>map("map"))
            .map(Map.Entry::getValue)
            .addTimestamps(Foo::getTimeMillisecond, LIMIT)
            .window(WindowDefinition.sliding(100, 10))
            .aggregate(FooAggregateOperations.aggregateFoo(),
                    (s, e, r) -> tuple3(s, e, r))
            .filter(t -> !isEmpty(t.f2()))
            .map(t -> String.format("started: %s\n%s\nended: %s\n", t.f0(), t.f2(), t.f1()))
            .drainTo(Sinks.files("sinkDirectory"));

Для этого нужно сохранить результат агрегирования во временном кортеже, а затем применить фильтрацию и затем окончательное сопоставление.

Я также создал проблему на GitHub , и мы рассмотрим поддержку этого поведения прямо в операции агрегирования.

...