Генерация данных на лету для бенчмаркинга Beam

Question

Моя цель - сравнить латентность и пропускную способность Apache Beam в сценарии использования потоковых данных с различными оконными запросами.

Я хочу создать свои собственные данные с помощью генератора данных на лету, чтобы вручную контролировать скорость генерации данных и использовать эти данные непосредственно из конвейера без механизма публикации / подписки, т.е. я не хочу читать данные от брокера и т. д., чтобы избежать узких мест. Есть ли способ сделать что-то похожее на то, что я хочу достичь? или есть ли исходный код для такого варианта использования с Beam SDK? Пока я не смог найти отправную точку, в существующих примерах кода используется механизм pub / sub, и они предполагают, что данные откуда-то приходят.

Спасибо за предложения заранее.

Answer 1

Что касается данных «на лету», одним из вариантов будет использование GenerateSequence, например:

pipeline.apply(GenerateSequence.from(0).withRate(RATE,Duration.millis(1000)))

Для создания других типов объектов вы можете использовать ParDo после использования Longи превратить его во что-то другое:

Pipeline p = Pipeline.create(PipelineOptionsFactory.create());
    p.apply(GenerateSequence.from(0).withRate(2, Duration.millis(1000)))
    .apply(Window.into(FixedWindows.of(Duration.standardSeconds(1))))
    .apply(FlatMapElements.into(TypeDescriptors.kvs(TypeDescriptors.strings(), TypeDescriptors.strings()))
        .via(i -> IntStream.range(0,2).mapToObj(k -> KV.of(String.format("Gen Value %s" , i),String.format("FlatMap Value %s ", k))).collect(Collectors.toList())))
    .apply(ParDo.of(new DoFn<KV<String,String>, String>() {
      @ProcessElement
      public void process(@Element KV<String,String> input){
        LOG.info("Value was {}", input);
      }
    }));
p.run();

Это должно генерировать такие значения, как:

Value was KV{Gen Value 0, FlatMap Value 0 }
Value was KV{Gen Value 0, FlatMap Value 1 }
Value was KV{Gen Value 1, FlatMap Value 0 }
Value was KV{Gen Value 1, FlatMap Value 1 }
Value was KV{Gen Value 2, FlatMap Value 0 }
Value was KV{Gen Value 2, FlatMap Value 1 }

Некоторые другие вещи, которые следует учитывать при тестировании производительности конвейеров:

• Прямой бегун предназначен для модульного тестирования, он делает классные вещи, такие как имитация сбоев, это помогает выявить проблемы, которые будут видны при запуске производственного конвейера.Однако он не предназначен для тестирования производительности.Я бы рекомендовал всегда использовать главный бегун для таких типов интеграционных тестов.

• Обратите внимание на оптимизацию Fusion Ссылка на документы , при использовании искусственного источника данных, такого как GenerateSequence, вам может понадобиться сделать GBK в качестве следующего шага, чтобы разрешитьработа должна быть распараллелена.Для бегуна потока данных больше информации можно найти здесь: Ссылка на документы

• В общем, для тестирования производительности, я бы рекомендовал тестировать весь сквозной конвейер.Существуют взаимодействия с источниками и приемниками (например, водяными знаками), которые не будут тестироваться в отдельном конвейере.

Надеюсь, что это поможет.