我FlinkKafkaConsumer011訂閱了一個主題。我希望apply在每個 kafka 消費者消息上處理 ( )，因此自定義在每個元素FooTrigger上返回TriggerResult.FIRE。以下代碼有效，我只是對timeWindowAll(Time.minutes(1)). 看起來我做錯了什么。// set up streaming execution environmentStreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();env.setStreamTimeCharacteristic(TimeCharacteristic.IngestionTime);// create a Kafka consumerFlinkKafkaConsumer011<Foo> consumer = new FlinkKafkaConsumer011<>( "topic", new Foo.FooSchema(), props); // Properties object// create Kafka consumer data sourceDataStream<FooTuple> trades = env.addSource(consumer) .timeWindowAll(Time.minutes(1)) .trigger(new FooTrigger()) .evictor(new FooEvictor()) .apply(new CreateFoos());
1 回答
當年話下
TA貢獻1890條經驗 獲得超9個贊
如果您的目標是將函數應用于流中的每個事件，ProcessFunction那么在 Flink 中使用a將是一種更自然的方法?；蛘咴诟唵蔚那闆r下，您可以使用地圖或平面地圖，或其豐富的變體，即 RichMapFunction 或 RichFlatMapFunction —— 這完全取決于您要嘗試做什么。
使用 map 或 flatmap，您可以執行無狀態的一對一或一對多轉換，它們的豐富變體可以使用鍵控狀態，而 ProcessFunction 可以使用狀態和計時器（前提是流已被鍵控）。
timeWindowAll 適用于流未按鍵分區的情況，并且您希望按持續時間定義的批處理進行非并行處理（對于鍵控并行窗口，請改用 timeWindow）。如果您只想在數據到達時對其進行處理，那么窗口化會增加不必要的復雜性。
添加回答
舉報
0/150
提交
取消