Flink окно перетянул поток производительности - PullRequest
Новая
       8

Flink окно перетянул поток производительности

0 голосов
/ 29 мая 2018

Я хотел бы понять, как окно влияет на производительность в задании Flink. Я выполняю задание flink, состоящее из оператора, как показано ниже:

Параллелизм задания (4,8,16): [источник автоматического создания]-> [Map1] -> [Tumbling Window (10s)] -> [Map2] -> [Sink]

Мгновенное оконное исполнение eps 4p, 8p, 16p

Задание, превышающее производительность, ограничено примерно 50k + - в секунду, независимо от того, как я масштабирую кластер с параллелизмом 4-16.

Когда окно задания было удалено, задание может достигать 200k в секунду.

Параллелизм заданий (4-8): [источник автоматического создания] -> [Map1] -> [Map2] -> [Sink]

Производительность Flink noокно 4p, 8p

Я удалил логику окна, чтобы исключить логику приложения, которая ограничивает производительность, но кажется, что окно все-таки приводило к снижению производительности всего моего потока, даже если это окно просто проходное.function.

Этот снимок экрана показывает, что производительность для задания без окна и задания с включенным пустым окном.

Пустое окно показывает всплеск, но оператор замедлен

Поток был ограничен и больше не мог масштабироваться.В любом случае, чтобы повысить производительность для такого сценария?

Примечание. Задание не имеет зависимостей от внешних служб и контрольная точка отключена.

1 Ответ

0 голосов
/ 02 июня 2018

Дальнейшие исследования показывают, что это вызвано keyBy () в операции keyedStream, производительность упала почти на 40%, когда поток был введен с помощью ключа, и если существует более одного keyedStream, он продолжает снижать производительность. 

        DataStream<Tuple2<String, MyObj>> testStream = finalizedStream
            .map(new RichMapFunction<MyObj, Tuple2<String, MyObj>>() {

                @Override
                public Tuple2<String, MyObj> map(MyObj value) throws Exception {
                    return new Tuple2<String, MyObj>(value.getCustomKey(), value);
                }
            });

    KeyedStream<Tuple2<String, MyObj>, Tuple> keyedStream = testStream.keyBy(0);
    keyedStream.map(new MapFunction<Tuple2<String, MyObj>, MyObj>() {

        @Override
        public MyObj map(Tuple2<String, MyObj> tuple) throws Exception {
            return tuple.f1;
        }
    }).name("keyed stream");

В любом случае, как я могу улучшить производительность?Кажется, использование ключа неизбежно.И я не уверен, почему мы пытались масштабировать работу и в нашей среде, производительность просто ограничивалась цифрой, независимо от того, сколько параллелизма мы обеспечили.Загрузка процессора и памяти показали, что он не был критическим, когда монитор использует nmon.

Любой вклад и помощь будут очень признательны.

...