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 13cfaa76b5eb90316ecb253d061f4b18b88fde27 Author: Weijie Guo <[email protected]> AuthorDate: Thu Mar 21 18:29:01 2024 +0800 [FLINK-34548][API] Introduce ProcessFunction and RuntimeContext related interfaces --- flink-datastream-api/pom.xml | 8 ++++ .../flink/datastream/api/common/Collector.java | 40 +++++++++++++++++ .../api/context/NonPartitionedContext.java | 35 +++++++++++++++ .../datastream/api/context/RuntimeContext.java | 33 ++++++++++++++ .../context/TwoOutputNonPartitionedContext.java | 36 ++++++++++++++++ .../api/function/ApplyPartitionFunction.java | 37 ++++++++++++++++ .../datastream/api/function/ProcessFunction.java | 50 ++++++++++++++++++++++ .../function/TwoOutputApplyPartitionFunction.java | 39 +++++++++++++++++ 8 files changed, 278 insertions(+) diff --git a/flink-datastream-api/pom.xml b/flink-datastream-api/pom.xml index f7fbbc35b1c..f5602c0ef89 100644 --- a/flink-datastream-api/pom.xml +++ b/flink-datastream-api/pom.xml @@ -33,4 +33,12 @@ under the License. <name>Flink : DataStream : API</name> <packaging>jar</packaging> + + <dependencies> + <dependency> + <groupId>org.apache.flink</groupId> + <artifactId>flink-core-api</artifactId> + <version>${project.version}</version> + </dependency> + </dependencies> </project> diff --git a/flink-datastream-api/src/main/java/org/apache/flink/datastream/api/common/Collector.java b/flink-datastream-api/src/main/java/org/apache/flink/datastream/api/common/Collector.java new file mode 100644 index 00000000000..5f2d25a6904 --- /dev/null +++ b/flink-datastream-api/src/main/java/org/apache/flink/datastream/api/common/Collector.java @@ -0,0 +1,40 @@ +/* + * 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.common; + +import org.apache.flink.annotation.Experimental; + +/** This class take response for collecting data to output stream. */ +@Experimental +public interface Collector<OUT> { + /** + * Collect record to output stream. + * + * @param record to be collected. + */ + void collect(OUT record); + + /** + * Overwrite the timestamp of this record and collect it to output stream. + * + * @param record to be collected. + * @param timestamp of the processed data. + */ + void collectAndOverwriteTimestamp(OUT record, long timestamp); +} diff --git a/flink-datastream-api/src/main/java/org/apache/flink/datastream/api/context/NonPartitionedContext.java b/flink-datastream-api/src/main/java/org/apache/flink/datastream/api/context/NonPartitionedContext.java new file mode 100644 index 00000000000..b96ae435232 --- /dev/null +++ b/flink-datastream-api/src/main/java/org/apache/flink/datastream/api/context/NonPartitionedContext.java @@ -0,0 +1,35 @@ +/* + * 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.context; + +import org.apache.flink.annotation.Experimental; +import org.apache.flink.datastream.api.function.ApplyPartitionFunction; + +/** + * This interface represents the context associated with all operations must be applied to all + * partitions. + */ +@Experimental +public interface NonPartitionedContext<OUT> extends RuntimeContext { + /** + * Apply a function to all partitions. For keyed stream, it will apply to all keys. For + * non-keyed stream, it will apply to single partition. + */ + void applyToAllPartitions(ApplyPartitionFunction<OUT> applyPartitionFunction) throws Exception; +} diff --git a/flink-datastream-api/src/main/java/org/apache/flink/datastream/api/context/RuntimeContext.java b/flink-datastream-api/src/main/java/org/apache/flink/datastream/api/context/RuntimeContext.java new file mode 100644 index 00000000000..4481417efd2 --- /dev/null +++ b/flink-datastream-api/src/main/java/org/apache/flink/datastream/api/context/RuntimeContext.java @@ -0,0 +1,33 @@ +/* + * 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.context; + +import org.apache.flink.annotation.Experimental; + +/** + * A RuntimeContext contains information about the context in which process functions are executed. + * Each parallel instance of the function will have a context through which it can access contextual + * information, such as the current key and execution mode. Through this context, we can also + * interact with the execution layer, such as getting state, emitting watermark, registering timer, + * etc. + */ +@Experimental +public interface RuntimeContext { + // TODO Introduce related methods in the subsequent RP. +} diff --git a/flink-datastream-api/src/main/java/org/apache/flink/datastream/api/context/TwoOutputNonPartitionedContext.java b/flink-datastream-api/src/main/java/org/apache/flink/datastream/api/context/TwoOutputNonPartitionedContext.java new file mode 100644 index 00000000000..283a37c6c9b --- /dev/null +++ b/flink-datastream-api/src/main/java/org/apache/flink/datastream/api/context/TwoOutputNonPartitionedContext.java @@ -0,0 +1,36 @@ +/* + * 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.context; + +import org.apache.flink.annotation.Experimental; +import org.apache.flink.datastream.api.function.TwoOutputApplyPartitionFunction; + +/** + * This interface represents the context associated with all operations must be applied to all + * partitions with two outputs. + */ +@Experimental +public interface TwoOutputNonPartitionedContext<OUT1, OUT2> extends RuntimeContext { + /** + * Apply a function to all partitions. For keyed stream, it will apply to all keys. For + * non-keyed stream, it will apply to single partition. + */ + void applyToAllPartitions(TwoOutputApplyPartitionFunction<OUT1, OUT2> applyPartitionFunction) + throws Exception; +} diff --git a/flink-datastream-api/src/main/java/org/apache/flink/datastream/api/function/ApplyPartitionFunction.java b/flink-datastream-api/src/main/java/org/apache/flink/datastream/api/function/ApplyPartitionFunction.java new file mode 100644 index 00000000000..d44846c8402 --- /dev/null +++ b/flink-datastream-api/src/main/java/org/apache/flink/datastream/api/function/ApplyPartitionFunction.java @@ -0,0 +1,37 @@ +/* + * 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.api.common.functions.Function; +import org.apache.flink.datastream.api.common.Collector; +import org.apache.flink.datastream.api.context.RuntimeContext; + +/** A function to be applied to all partitions . */ +@FunctionalInterface +@Experimental +public interface ApplyPartitionFunction<OUT> extends Function { + /** + * The actual method to be applied to each partition. + * + * @param collector to output data. + * @param ctx runtime context in which this function is executed. + */ + void apply(Collector<OUT> collector, RuntimeContext ctx) throws Exception; +} diff --git a/flink-datastream-api/src/main/java/org/apache/flink/datastream/api/function/ProcessFunction.java b/flink-datastream-api/src/main/java/org/apache/flink/datastream/api/function/ProcessFunction.java new file mode 100644 index 00000000000..2efee819aff --- /dev/null +++ b/flink-datastream-api/src/main/java/org/apache/flink/datastream/api/function/ProcessFunction.java @@ -0,0 +1,50 @@ +/* + * 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.api.common.functions.Function; + +/** Base class for all user defined process functions. */ +@Experimental +public interface ProcessFunction extends Function { + /** + * Initialization method for the function. It is called before the actual working methods (like + * processRecord) and thus suitable for one time setup work. + * + * <p>By default, this method does nothing. + * + * @throws Exception Implementations may forward exceptions, which are caught by the runtime. + * When the runtime catches an exception, it aborts the task and lets the fail-over logic + * decide whether to retry the task execution. + */ + default void open() throws Exception {} + + /** + * Tear-down method for the user code. It is called after the last call to the main working + * methods (e.g. processRecord). + * + * <p>This method can be used for clean up work. + * + * @throws Exception Implementations may forward exceptions, which are caught by the runtime. + * When the runtime catches an exception, it aborts the task and lets the fail-over logic + * decide whether to retry the task execution. + */ + default void close() throws Exception {} +} diff --git a/flink-datastream-api/src/main/java/org/apache/flink/datastream/api/function/TwoOutputApplyPartitionFunction.java b/flink-datastream-api/src/main/java/org/apache/flink/datastream/api/function/TwoOutputApplyPartitionFunction.java new file mode 100644 index 00000000000..786d79ba9ca --- /dev/null +++ b/flink-datastream-api/src/main/java/org/apache/flink/datastream/api/function/TwoOutputApplyPartitionFunction.java @@ -0,0 +1,39 @@ +/* + * 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.api.common.functions.Function; +import org.apache.flink.datastream.api.common.Collector; +import org.apache.flink.datastream.api.context.RuntimeContext; + +/** A function to be applied to all partitions with two outputs. */ +@FunctionalInterface +@Experimental +public interface TwoOutputApplyPartitionFunction<OUT1, OUT2> extends Function { + /** + * The actual method to be applied to each partition. + * + * @param firstOutput to emit record to first output. + * @param secondOutput to emit record to second output. + * @param ctx runtime context in which this function is executed. + */ + void apply(Collector<OUT1> firstOutput, Collector<OUT2> secondOutput, RuntimeContext ctx) + throws Exception; +}
