This is an automated email from the ASF dual-hosted git repository. guoweijie pushed a commit to branch master in repository https://gitbox.apache.org/repos/asf/flink.git
commit cedbcce6eff6ae6201e192b9f5c593efff5fc977 Author: Weijie Guo <[email protected]> AuthorDate: Tue Feb 20 15:18:18 2024 +0800 [FLINK-34548][API] Introduce variants of ProcessFunction --- .../function/OneInputStreamProcessFunction.java | 45 ++++++++++++++ .../TwoInputBroadcastStreamProcessFunction.java | 68 ++++++++++++++++++++++ .../TwoInputNonBroadcastStreamProcessFunction.java | 64 ++++++++++++++++++++ .../function/TwoOutputStreamProcessFunction.java | 48 +++++++++++++++ 4 files changed, 225 insertions(+) diff --git a/flink-datastream-api/src/main/java/org/apache/flink/datastream/api/function/OneInputStreamProcessFunction.java b/flink-datastream-api/src/main/java/org/apache/flink/datastream/api/function/OneInputStreamProcessFunction.java new file mode 100644 index 00000000000..d34eaa2daeb --- /dev/null +++ b/flink-datastream-api/src/main/java/org/apache/flink/datastream/api/function/OneInputStreamProcessFunction.java @@ -0,0 +1,45 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.flink.datastream.api.function; + +import org.apache.flink.annotation.Experimental; +import org.apache.flink.datastream.api.common.Collector; +import org.apache.flink.datastream.api.context.NonPartitionedContext; +import org.apache.flink.datastream.api.context.RuntimeContext; + +/** This contains all logical related to process records from single input. */ +@Experimental +public interface OneInputStreamProcessFunction<IN, OUT> extends ProcessFunction { + /** + * Process record and emit data through {@link Collector}. + * + * @param record to process. + * @param output to emit processed records. + * @param ctx runtime context in which this function is executed. + */ + void processRecord(IN record, Collector<OUT> output, RuntimeContext ctx) throws Exception; + + /** + * This is a life-cycle method indicates that this function will no longer receive any data from + * the input. + * + * @param ctx the context in which this function is executed. + */ + default void endInput(NonPartitionedContext<OUT> ctx) {} +} diff --git a/flink-datastream-api/src/main/java/org/apache/flink/datastream/api/function/TwoInputBroadcastStreamProcessFunction.java b/flink-datastream-api/src/main/java/org/apache/flink/datastream/api/function/TwoInputBroadcastStreamProcessFunction.java new file mode 100644 index 00000000000..881e0496349 --- /dev/null +++ b/flink-datastream-api/src/main/java/org/apache/flink/datastream/api/function/TwoInputBroadcastStreamProcessFunction.java @@ -0,0 +1,68 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.flink.datastream.api.function; + +import org.apache.flink.annotation.Experimental; +import org.apache.flink.datastream.api.common.Collector; +import org.apache.flink.datastream.api.context.NonPartitionedContext; +import org.apache.flink.datastream.api.context.RuntimeContext; + +/** + * This contains all logical related to process records from a broadcast stream and a non-broadcast + * stream. + */ +@Experimental +public interface TwoInputBroadcastStreamProcessFunction<IN1, IN2, OUT> extends ProcessFunction { + /** + * Process record from non-broadcast input and emit data through {@link Collector}. + * + * @param record to process. + * @param output to emit processed records. + * @param ctx runtime context in which this function is executed. + */ + void processRecordFromNonBroadcastInput(IN1 record, Collector<OUT> output, RuntimeContext ctx) + throws Exception; + + /** + * Process record from broadcast input. In general, the broadcast side is not allowed to + * manipulate state and output data because it corresponds to all partitions instead of a single + * partition. But you could use broadcast context to process all the partitions at once. + * + * @param record to process. + * @param ctx the context in which this function is executed. + */ + void processRecordFromBroadcastInput(IN2 record, NonPartitionedContext<OUT> ctx) + throws Exception; + + /** + * This is a life-cycle method indicates that this function will no longer receive any data from + * the non-broadcast input. + * + * @param ctx the context in which this function is executed. + */ + default void endNonBroadcastInput(NonPartitionedContext<OUT> ctx) {} + + /** + * This is a life-cycle method indicates that this function will no longer receive any data from + * the broadcast input. + * + * @param ctx the context in which this function is executed. + */ + default void endBroadcastInput(NonPartitionedContext<OUT> ctx) {} +} diff --git a/flink-datastream-api/src/main/java/org/apache/flink/datastream/api/function/TwoInputNonBroadcastStreamProcessFunction.java b/flink-datastream-api/src/main/java/org/apache/flink/datastream/api/function/TwoInputNonBroadcastStreamProcessFunction.java new file mode 100644 index 00000000000..937ca73f69e --- /dev/null +++ b/flink-datastream-api/src/main/java/org/apache/flink/datastream/api/function/TwoInputNonBroadcastStreamProcessFunction.java @@ -0,0 +1,64 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.flink.datastream.api.function; + +import org.apache.flink.annotation.Experimental; +import org.apache.flink.datastream.api.common.Collector; +import org.apache.flink.datastream.api.context.NonPartitionedContext; +import org.apache.flink.datastream.api.context.RuntimeContext; + +/** This contains all logical related to process records from two non-broadcast input. */ +@Experimental +public interface TwoInputNonBroadcastStreamProcessFunction<IN1, IN2, OUT> extends ProcessFunction { + /** + * Process record from the first input and emit data through {@link Collector}. + * + * @param record to process. + * @param output to emit processed records. + * @param ctx runtime context in which this function is executed. + */ + void processRecordFromFirstInput(IN1 record, Collector<OUT> output, RuntimeContext ctx) + throws Exception; + + /** + * Process record from the second input and emit data through {@link Collector}. + * + * @param record to process. + * @param output to emit processed records. + * @param ctx runtime context in which this function is executed. + */ + void processRecordFromSecondInput(IN2 record, Collector<OUT> output, RuntimeContext ctx) + throws Exception; + + /** + * This is a life-cycle method indicates that this function will no longer receive any data from + * the first input. + * + * @param ctx the context in which this function is executed. + */ + default void endFirstInput(NonPartitionedContext<OUT> ctx) {} + + /** + * This is a life-cycle method indicates that this function will no longer receive any data from + * the second input. + * + * @param ctx the context in which this function is executed. + */ + default void endSecondInput(NonPartitionedContext<OUT> ctx) {} +} diff --git a/flink-datastream-api/src/main/java/org/apache/flink/datastream/api/function/TwoOutputStreamProcessFunction.java b/flink-datastream-api/src/main/java/org/apache/flink/datastream/api/function/TwoOutputStreamProcessFunction.java new file mode 100644 index 00000000000..225656f4e84 --- /dev/null +++ b/flink-datastream-api/src/main/java/org/apache/flink/datastream/api/function/TwoOutputStreamProcessFunction.java @@ -0,0 +1,48 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.flink.datastream.api.function; + +import org.apache.flink.annotation.Experimental; +import org.apache.flink.datastream.api.common.Collector; +import org.apache.flink.datastream.api.context.RuntimeContext; +import org.apache.flink.datastream.api.context.TwoOutputNonPartitionedContext; + +/** This contains all logical related to process and emit records to two output streams. */ +@Experimental +public interface TwoOutputStreamProcessFunction<IN, OUT1, OUT2> extends ProcessFunction { + /** + * Process and emit record to the first/second output through {@link Collector}s. + * + * @param record to process. + * @param output1 to emit processed records to the first output. + * @param output2 to emit processed records to the second output. + * @param ctx runtime context in which this function is executed. + */ + void processRecord( + IN record, Collector<OUT1> output1, Collector<OUT2> output2, RuntimeContext ctx) + throws Exception; + + /** + * This is a life-cycle method indicates that this function will no longer receive any input + * data. + * + * @param ctx the context in which this function is executed. + */ + default void endInput(TwoOutputNonPartitionedContext<OUT1, OUT2> ctx) {} +}
