Применить на каждом входе

Question

У меня есть FlinkKafkaConsumer011 подписался на тему.Я хочу обработать (apply) для каждого сообщения потребителя kafka, следовательно, custom FooTrigger, который возвращает TriggerResult.FIRE для каждого элемента.

Следующий код работает, я просто запутался в timeWindowAll(Time.minutes(1)).Похоже, я делаю что-то не так.

// set up streaming execution environment
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
env.setStreamTimeCharacteristic(TimeCharacteristic.IngestionTime);

// create a Kafka consumer
FlinkKafkaConsumer011<Foo> consumer =  new FlinkKafkaConsumer011<>(
    "topic",
    new Foo.FooSchema(),
    props);   // Properties object

// create Kafka consumer data source
DataStream<FooTuple> trades = env.addSource(consumer)
    .timeWindowAll(Time.minutes(1))
    .trigger(new FooTrigger())
    .evictor(new FooEvictor())
    .apply(new CreateFoos());

Answer 1

Если ваша цель - применить функцию к каждому событию в потоке, ProcessFunction будет более естественным способом сделать это в Flink.Или, в более простых случаях, вы можете использовать карту или плоскую карту или их расширенные варианты, например RichMapFunction или RichFlatMapFunction - все зависит от того, что вы пытаетесь сделать.

С картой или плоской картой вы можетевыполнить преобразование «один-к-одному» или «один-ко-многим» без сохранения состояния, их расширенные варианты могут использовать состояние ключа, а ProcessFunction может использовать состояние и таймеры (при условии, что для потока был задан ключ).

timeWindowAll предназначен дляслучаи, когда поток не разделен ключом, и вы хотите выполнять непараллельную обработку в пакетах, определяемых продолжительностью времени (для параллельного оконного управления с ключами вместо этого используйте timeWindow).Если все, что вам нужно, - это обрабатывать данные по мере их поступления, тогда создание окон добавляет ненужную сложность.