Используя окно Flink и функцию сгиба, элемент отсутствует?

Question

Когда я пытаюсь агрегировать элементы, используя функцию окна и сворачивания, некоторые из элементов упускаются из-за агрегации. Расходные элементы от Кафки (value:0, value:1, value:2, value:3) и объединяя их в нечетные и четные значения.

Выход:

{even=[0, 2, 4], odd=[1, 3]}
{even=[6, 8], odd=[5, 7, 9]}
{even=[14, 16, 18], odd=[15, 17]}
{even=[20, 22], odd=[19, 21, 23]}
{even=[24, 26, 28], odd=[25, 27]}

Числа между 10-13 отсутствуют, и это происходит для случайного набора чисел. Может кто-нибудь подсказать, что пропущено из приведенного ниже кода и как я могу быть уверен, что обработал все элементы?

public static class Splitter implements FlatMapFunction<String, 
    Tuple3<String, String, List<String>>{
    private static final long serialVersionUID = 1L;

    @Override
    public void flatMap(String value, Collector<Tuple3<String, String, 
        List<String>>out) throws Exception {
        String[] vals = value.split(":");

        if(vals.length 1 && Integer.parseInt(vals[1]) % 2 == 0){
            out.collect(new Tuple3<String, String, List<String>>
             ("test","even", Arrays.asList(vals[1])));
        }else{
            out.collect(new Tuple3<String, String, List<String>>
            ("test","odd", Arrays.asList(vals[1])));
        }
    }
}


    DataStream<Map<String, List<String>>streamValue = 
    kafkaStream.flatMap(new Splitter()).keyBy(0)
    .window(TumblingEventTimeWindows.of(Time.milliseconds(3000))).
    trigger(CustomizedCountTrigger.of(5L))//.trigger(CountTrigger.of(2))
    .fold(new HashMap<String, List<String>>(), new 
    FoldFunction<Tuple3<String, String, List<String>>, Map<String, 
    List<String>>>() {
        private static final long serialVersionUID = 1L;

        @Override
        public Map<String, List<String>fold(Map<String, 
        List<String>accumulator,
        Tuple3<String, String, List<String>value) throws 
        Exception {
            if(accumulator.get(value.f1) != null){
                List<Stringlist = new ArrayList<>();
                list.addAll(accumulator.get(value.f1));
                list.addAll(value.f2);
                accumulator.put(value.f1, list);
            }else{
                accumulator.put(value.f1, value.f2);
            }
            return accumulator;
        }
    });

    streamValue.print();
    env.execute("window test");
}


public class CustomizedCountTrigger<W extends Windowextends 
Trigger<Object, W{

    private static final long serialVersionUID = 1L;
    private final long maxCount;

    private final ReducingStateDescriptor<LongstateDesc =
    new ReducingStateDescriptor<>("count", new Sum(), 
    LongSerializer.INSTANCE);

    private CustomizedCountTrigger(long maxCount) {
        this.maxCount = maxCount;
    }

    @Override
    public TriggerResult onElement(Object element, long timestamp, W window,
    TriggerContext ctx) throws Exception {
        ReducingState<Longcount = ctx.getPartitionedState(stateDesc);
        count.add(1L);
        if (count.get() >= maxCount) {
            count.clear();
            return TriggerResult.FIRE_AND_PURGE;
        }
        return TriggerResult.CONTINUE;
    }

    @Override
    public TriggerResult onProcessingTime(long time, W window,

    org.apache.flink.streaming.api.windowing.triggers.Trigger.TriggerContext

    ctx) throws Exception {
        return TriggerResult.CONTINUE;
    }

    @Override
    public TriggerResult onEventTime(long time, W window,

    org.apache.flink.streaming.api.windowing.triggers.Trigger.TriggerContext

    ctx) throws Exception {
        return TriggerResult.CONTINUE;
    }

    @Override
    public void clear(W window, 
    org.apache.flink.streaming.api.windowing.triggers.Trigger.TriggerContext

    ctx)
    throws Exception {
        ctx.getPartitionedState(stateDesc).clear();
    }

    @Override
    public String toString() {
        return "CountTrigger(" +  maxCount + ")";
    }

    public static <W extends WindowCustomizedCountTrigger<Wof(long 
    maxCount) {
        return new CustomizedCountTrigger<>(maxCount);
    }

    private static class Sum implements ReduceFunction<Long{
        private static final long serialVersionUID = 1L;

        @Override
        public Long reduce(Long value1, Long value2) throws Exception {
            return value1 + value2;
        }

    }
}

Answer 1

Поэтому я начал писать первую часть этого, прежде чем заметил, что ваши пользовательские триггеры делают тот факт, что вы используете окно TumblingEventTime, как-то неуместно, но я все равно хочу включить мои исходные мысли, так как я не до конца уверен, почему вы используете окно EventTime. когда вы не используете его. Мой ответ после осознания этого ниже оригинала.

Вы запускаете это на одном параллельном или множественном? Причина, по которой я спрашиваю, заключается в том, что если это множественный параллелизм (и тема kafka также состоит из нескольких разделов), то возможно, что сообщения принимаются и обрабатываются в непоследовательном порядке. Это может привести к сообщениям с отметкой времени, что приводит к продвижению водяного знака, в результате чего окно обрабатывает сообщения. Затем у следующего (ых) сообщения (й) будет время события, которое предшествует текущему времени водяного знака (или "опоздание"), что приведет к удалению сообщения.

Так, например: если у вас есть 20 элементов, и время каждого события примерно такое:

message1: eventTime: 1000 message1: eventTime: 2000 и т. д...

И ваше временное окно составляет 5001мс.

Теперь сообщения с сообщений 1 по 9 проходят по порядку. Это первое окно будет обработано и будет содержать сообщения 1-5 (message6 вызовет обработку окна). Теперь, если сообщение 11 поступает до сообщения 10, это приведет к обработке окна, содержащего сообщения 6-9. И когда появляется сообщение 10, водяной знак уже опередил время события сообщения 10, в результате чего он был отброшен как "позднее событие".

Правильный ответ

Вместо использования окна eventTime и пользовательского триггера, попробуйте использовать countWindow.

Так что замените это:

.window(TumblingEventTimeWindows.of(Time.milliseconds(3000))).
trigger(CustomizedCountTrigger.of(5L))//.trigger(CountTrigger.of(2))

С этим:

.countWindow(5L)