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 9fa74a8a7061af3c518853fc10bfa7d7abf5ea6d Author: Weijie Guo <[email protected]> AuthorDate: Thu Mar 21 18:39:50 2024 +0800 [FLINK-34548][API] Introduce stream interface and move KeySelector to flink-core-api --- .../flink/api/java/functions/KeySelector.java | 0 .../datastream/api/stream/BroadcastStream.java | 76 ++++++++ .../flink/datastream/api/stream/DataStream.java | 25 +++ .../flink/datastream/api/stream/GlobalStream.java | 95 ++++++++++ .../api/stream/KeyedPartitionStream.java | 211 +++++++++++++++++++++ .../api/stream/NonKeyedPartitionStream.java | 117 ++++++++++++ pom.xml | 1 + 7 files changed, 525 insertions(+) diff --git a/flink-core/src/main/java/org/apache/flink/api/java/functions/KeySelector.java b/flink-core-api/src/main/java/org/apache/flink/api/java/functions/KeySelector.java similarity index 100% rename from flink-core/src/main/java/org/apache/flink/api/java/functions/KeySelector.java rename to flink-core-api/src/main/java/org/apache/flink/api/java/functions/KeySelector.java diff --git a/flink-datastream-api/src/main/java/org/apache/flink/datastream/api/stream/BroadcastStream.java b/flink-datastream-api/src/main/java/org/apache/flink/datastream/api/stream/BroadcastStream.java new file mode 100644 index 00000000000..f8856df2326 --- /dev/null +++ b/flink-datastream-api/src/main/java/org/apache/flink/datastream/api/stream/BroadcastStream.java @@ -0,0 +1,76 @@ +/* + * 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.stream; + +import org.apache.flink.annotation.Experimental; +import org.apache.flink.api.java.functions.KeySelector; +import org.apache.flink.datastream.api.function.TwoInputBroadcastStreamProcessFunction; + +/** This interface represents a stream that each parallel task processes the same data. */ +@Experimental +public interface BroadcastStream<T> extends DataStream { + /** + * Apply a two input operation to this and other {@link KeyedPartitionStream}. + * + * <p>Generally, concatenating {@link BroadcastStream} and {@link KeyedPartitionStream} will + * result in a {@link NonKeyedPartitionStream}, and you can manually generate a {@link + * KeyedPartitionStream} via keyBy partitioning. In some cases, you can guarantee that the + * partition on which the data is processed will not change, then you can use {@link + * #connectAndProcess(KeyedPartitionStream, TwoInputBroadcastStreamProcessFunction, + * KeySelector)} to avoid shuffling. + * + * @param other {@link KeyedPartitionStream} to perform operation with two input. + * @param processFunction to perform operation. + * @return new stream with this operation. + */ + <K, T_OTHER, OUT> NonKeyedPartitionStream<OUT> connectAndProcess( + KeyedPartitionStream<K, T_OTHER> other, + TwoInputBroadcastStreamProcessFunction<T_OTHER, T, OUT> processFunction); + + /** + * Apply a two input operation to this and other {@link NonKeyedPartitionStream}. + * + * @param other {@link NonKeyedPartitionStream} to perform operation with two input. + * @param processFunction to perform operation. + * @return new stream with this operation. + */ + <T_OTHER, OUT> NonKeyedPartitionStream<OUT> connectAndProcess( + NonKeyedPartitionStream<T_OTHER> other, + TwoInputBroadcastStreamProcessFunction<T_OTHER, T, OUT> processFunction); + + /** + * Apply a two input operation to this and other {@link KeyedPartitionStream}. + * + * <p>This method is used to avoid shuffle after applying the process function. It is required + * that for the record from non-broadcast input, the new {@link KeySelector} must extract the + * same key as the original {@link KeySelector}s on the {@link KeyedPartitionStream}. Otherwise, + * the partition of data will be messy. As for the record from broadcast input, the output key + * from keyed partition itself instead of the new key selector, so the data it outputs will not + * affect the partition. + * + * @param other {@link KeyedPartitionStream} to perform operation with two input. + * @param processFunction to perform operation. + * @param newKeySelector to select the key after process. + * @return new {@link KeyedPartitionStream} with this operation. + */ + <K, T_OTHER, OUT> KeyedPartitionStream<K, OUT> connectAndProcess( + KeyedPartitionStream<K, T_OTHER> other, + TwoInputBroadcastStreamProcessFunction<T_OTHER, T, OUT> processFunction, + KeySelector<OUT, K> newKeySelector); +} diff --git a/flink-datastream-api/src/main/java/org/apache/flink/datastream/api/stream/DataStream.java b/flink-datastream-api/src/main/java/org/apache/flink/datastream/api/stream/DataStream.java new file mode 100644 index 00000000000..c782b374c40 --- /dev/null +++ b/flink-datastream-api/src/main/java/org/apache/flink/datastream/api/stream/DataStream.java @@ -0,0 +1,25 @@ +/* + * 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.stream; + +import org.apache.flink.annotation.Experimental; + +/** This is the topmost base interface of all streams of DataStream V2 API. */ +@Experimental +public interface DataStream {} diff --git a/flink-datastream-api/src/main/java/org/apache/flink/datastream/api/stream/GlobalStream.java b/flink-datastream-api/src/main/java/org/apache/flink/datastream/api/stream/GlobalStream.java new file mode 100644 index 00000000000..cb8e0b83dd7 --- /dev/null +++ b/flink-datastream-api/src/main/java/org/apache/flink/datastream/api/stream/GlobalStream.java @@ -0,0 +1,95 @@ +/* + * 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.stream; + +import org.apache.flink.annotation.Experimental; +import org.apache.flink.api.java.functions.KeySelector; +import org.apache.flink.datastream.api.function.OneInputStreamProcessFunction; +import org.apache.flink.datastream.api.function.TwoInputNonBroadcastStreamProcessFunction; +import org.apache.flink.datastream.api.function.TwoOutputStreamProcessFunction; + +/** This interface represents a stream that force single parallelism. */ +@Experimental +public interface GlobalStream<T> extends DataStream { + /** + * Apply an operation to this {@link GlobalStream}. + * + * @param processFunction to perform operation. + * @return new stream with this operation. + */ + <OUT> GlobalStream<OUT> process(OneInputStreamProcessFunction<T, OUT> processFunction); + + /** + * Apply a two output operation to this {@link GlobalStream}. + * + * @param processFunction to perform two output operation. + * @return new stream with this operation. + */ + <OUT1, OUT2> TwoGlobalStreams<OUT1, OUT2> process( + TwoOutputStreamProcessFunction<T, OUT1, OUT2> processFunction); + + /** + * Apply a two input operation to this and other {@link GlobalStream}. + * + * @param other {@link GlobalStream} to perform operation with two input. + * @param processFunction to perform operation. + * @return new stream with this operation. + */ + <T_OTHER, OUT> GlobalStream<OUT> connectAndProcess( + GlobalStream<T_OTHER> other, + TwoInputNonBroadcastStreamProcessFunction<T, T_OTHER, OUT> processFunction); + + /** + * Transform this stream to a {@link KeyedPartitionStream}. + * + * @param keySelector to decide how to map data to partition. + * @return the transformed stream partitioned by key. + */ + <K> KeyedPartitionStream<K, T> keyBy(KeySelector<T, K> keySelector); + + /** + * Transform this stream to a new {@link NonKeyedPartitionStream}, data will be shuffled between + * these two streams. + * + * @return the transformed stream after shuffle. + */ + NonKeyedPartitionStream<T> shuffle(); + + /** + * Transform this stream to a new {@link BroadcastStream}. + * + * @return the transformed {@link BroadcastStream}. + */ + BroadcastStream<T> broadcast(); + + // TODO add toSink method. + + /** + * This class represents a combination of two {@link GlobalStream}. It will be used as the + * return value of operation with two output. + */ + @Experimental + interface TwoGlobalStreams<T1, T2> { + /** Get the first stream. */ + GlobalStream<T1> getFirst(); + + /** Get the second stream. */ + GlobalStream<T2> getSecond(); + } +} diff --git a/flink-datastream-api/src/main/java/org/apache/flink/datastream/api/stream/KeyedPartitionStream.java b/flink-datastream-api/src/main/java/org/apache/flink/datastream/api/stream/KeyedPartitionStream.java new file mode 100644 index 00000000000..40e15f08646 --- /dev/null +++ b/flink-datastream-api/src/main/java/org/apache/flink/datastream/api/stream/KeyedPartitionStream.java @@ -0,0 +1,211 @@ +/* + * 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.stream; + +import org.apache.flink.annotation.Experimental; +import org.apache.flink.api.java.functions.KeySelector; +import org.apache.flink.datastream.api.function.OneInputStreamProcessFunction; +import org.apache.flink.datastream.api.function.TwoInputBroadcastStreamProcessFunction; +import org.apache.flink.datastream.api.function.TwoInputNonBroadcastStreamProcessFunction; +import org.apache.flink.datastream.api.function.TwoOutputStreamProcessFunction; +import org.apache.flink.datastream.api.stream.NonKeyedPartitionStream.TwoNonKeyedPartitionStreams; + +/** + * This interface represents a kind of partitioned data stream. For this stream, each key is a + * partition, and the partition to which the data belongs is deterministic. + */ +@Experimental +public interface KeyedPartitionStream<K, T> extends DataStream { + /** + * Apply an operation to this {@link KeyedPartitionStream}. + * + * <p>This method is used to avoid shuffle after applying the process function. It is required + * that for the same record, the new {@link KeySelector} must extract the same key as the + * original {@link KeySelector} on this {@link KeyedPartitionStream}. Otherwise, the partition + * of data will be messy. + * + * @param processFunction to perform operation. + * @param newKeySelector to select the key after process. + * @return new {@link KeyedPartitionStream} with this operation. + */ + <OUT> KeyedPartitionStream<K, OUT> process( + OneInputStreamProcessFunction<T, OUT> processFunction, + KeySelector<OUT, K> newKeySelector); + + /** + * Apply an operation to this {@link KeyedPartitionStream}. + * + * <p>Generally, apply an operation to a {@link KeyedPartitionStream} will result in a {@link + * NonKeyedPartitionStream}, and you can manually generate a {@link KeyedPartitionStream} via + * keyBy partitioning. In some cases, you can guarantee that the partition on which the data is + * processed will not change, then you can use {@link #process(OneInputStreamProcessFunction, + * KeySelector)} to avoid shuffling. + * + * @param processFunction to perform operation. + * @return new {@link NonKeyedPartitionStream} with this operation. + */ + <OUT> NonKeyedPartitionStream<OUT> process( + OneInputStreamProcessFunction<T, OUT> processFunction); + + /** + * Apply a two output operation to this {@link KeyedPartitionStream}. + * + * <p>This method is used to avoid shuffle after applying the process function. It is required + * that for the same record, these new two {@link KeySelector}s must extract the same key as the + * original {@link KeySelector}s on this {@link KeyedPartitionStream}. Otherwise, the partition + * of data will be messy. + * + * @param processFunction to perform two output operation. + * @param keySelector1 to select the key of first output. + * @param keySelector2 to select the key of second output. + * @return new {@link TwoKeyedPartitionStreams} with this operation. + */ + <OUT1, OUT2> TwoKeyedPartitionStreams<K, OUT1, OUT2> process( + TwoOutputStreamProcessFunction<T, OUT1, OUT2> processFunction, + KeySelector<OUT1, K> keySelector1, + KeySelector<OUT2, K> keySelector2); + + /** + * Apply a two output operation to this {@link KeyedPartitionStream}. + * + * @param processFunction to perform two output operation. + * @return new {@link TwoNonKeyedPartitionStreams} with this operation. + */ + <OUT1, OUT2> TwoNonKeyedPartitionStreams<OUT1, OUT2> process( + TwoOutputStreamProcessFunction<T, OUT1, OUT2> processFunction); + + /** + * Apply a two input operation to this and other {@link KeyedPartitionStream}. The two keyed + * streams must have the same partitions, otherwise it makes no sense to connect them. + * + * <p>Generally, concatenating two {@link KeyedPartitionStream} will result in a {@link + * NonKeyedPartitionStream}, and you can manually generate a {@link KeyedPartitionStream} via + * keyBy partitioning. In some cases, you can guarantee that the partition on which the data is + * processed will not change, then you can use {@link #connectAndProcess(KeyedPartitionStream, + * TwoInputNonBroadcastStreamProcessFunction, KeySelector)} to avoid shuffling. + * + * @param other {@link KeyedPartitionStream} to perform operation with two input. + * @param processFunction to perform operation. + * @return new {@link NonKeyedPartitionStream} with this operation. + */ + <T_OTHER, OUT> NonKeyedPartitionStream<OUT> connectAndProcess( + KeyedPartitionStream<K, T_OTHER> other, + TwoInputNonBroadcastStreamProcessFunction<T, T_OTHER, OUT> processFunction); + + /** + * Apply a two input operation to this and other {@link KeyedPartitionStream}.The two keyed + * streams must have the same partitions, otherwise it makes no sense to connect them. + * + * <p>This method is used to avoid shuffle after applying the process function. It is required + * that for the same record, the new {@link KeySelector} must extract the same key as the + * original {@link KeySelector}s on these two {@link KeyedPartitionStream}s. Otherwise, the + * partition of data will be messy. + * + * @param other {@link KeyedPartitionStream} to perform operation with two input. + * @param processFunction to perform operation. + * @param newKeySelector to select the key after process. + * @return new {@link KeyedPartitionStream} with this operation. + */ + <T_OTHER, OUT> KeyedPartitionStream<K, OUT> connectAndProcess( + KeyedPartitionStream<K, T_OTHER> other, + TwoInputNonBroadcastStreamProcessFunction<T, T_OTHER, OUT> processFunction, + KeySelector<OUT, K> newKeySelector); + + /** + * Apply a two input operation to this and other {@link BroadcastStream}. + * + * <p>Generally, concatenating {@link KeyedPartitionStream} and {@link BroadcastStream} will + * result in a {@link NonKeyedPartitionStream}, and you can manually generate a {@link + * KeyedPartitionStream} via keyBy partitioning. In some cases, you can guarantee that the + * partition on which the data is processed will not change, then you can use {@link + * #connectAndProcess(BroadcastStream, TwoInputBroadcastStreamProcessFunction, KeySelector)} to + * avoid shuffling. + * + * @param processFunction to perform operation. + * @return new stream with this operation. + */ + <T_OTHER, OUT> NonKeyedPartitionStream<OUT> connectAndProcess( + BroadcastStream<T_OTHER> other, + TwoInputBroadcastStreamProcessFunction<T, T_OTHER, OUT> processFunction); + + /** + * Apply a two input operation to this and other {@link BroadcastStream}. + * + * <p>This method is used to avoid shuffle after applying the process function. It is required + * that for the record from non-broadcast input, the new {@link KeySelector} must extract the + * same key as the original {@link KeySelector}s on the {@link KeyedPartitionStream}. Otherwise, + * the partition of data will be messy. As for the record from broadcast input, the output key + * from keyed partition itself instead of the new key selector, so the data it outputs will not + * affect the partition. + * + * @param other {@link BroadcastStream} to perform operation with two input. + * @param processFunction to perform operation. + * @param newKeySelector to select the key after process. + * @return new {@link KeyedPartitionStream} with this operation. + */ + <T_OTHER, OUT> KeyedPartitionStream<K, OUT> connectAndProcess( + BroadcastStream<T_OTHER> other, + TwoInputBroadcastStreamProcessFunction<T, T_OTHER, OUT> processFunction, + KeySelector<OUT, K> newKeySelector); + + /** + * Coalesce this stream to a {@link GlobalStream}. + * + * @return the coalesced global stream. + */ + GlobalStream<T> global(); + + /** + * Transform this stream to a new {@link KeyedPartitionStream}. + * + * @param keySelector to decide how to map data to partition. + * @return the transformed stream partitioned by key. + */ + <NEW_KEY> KeyedPartitionStream<NEW_KEY, T> keyBy(KeySelector<T, NEW_KEY> keySelector); + + /** + * Transform this stream to a new {@link NonKeyedPartitionStream}, data will be shuffled between + * these two streams. + * + * @return the transformed stream after shuffle. + */ + NonKeyedPartitionStream<T> shuffle(); + + /** + * Transform this stream to a new {@link BroadcastStream}. + * + * @return the transformed {@link BroadcastStream}. + */ + BroadcastStream<T> broadcast(); + + // TODO add toSink method. + + /** + * This class represents a combination of two {@link KeyedPartitionStream}. It will be used as + * the return value of operation with two output. + */ + @Experimental + interface TwoKeyedPartitionStreams<K, T1, T2> { + /** Get the first stream. */ + KeyedPartitionStream<K, T1> getFirst(); + + /** Get the second stream. */ + KeyedPartitionStream<K, T2> getSecond(); + } +} diff --git a/flink-datastream-api/src/main/java/org/apache/flink/datastream/api/stream/NonKeyedPartitionStream.java b/flink-datastream-api/src/main/java/org/apache/flink/datastream/api/stream/NonKeyedPartitionStream.java new file mode 100644 index 00000000000..dd4a1ee2855 --- /dev/null +++ b/flink-datastream-api/src/main/java/org/apache/flink/datastream/api/stream/NonKeyedPartitionStream.java @@ -0,0 +1,117 @@ +/* + * 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.stream; + +import org.apache.flink.annotation.Experimental; +import org.apache.flink.api.java.functions.KeySelector; +import org.apache.flink.datastream.api.function.OneInputStreamProcessFunction; +import org.apache.flink.datastream.api.function.TwoInputBroadcastStreamProcessFunction; +import org.apache.flink.datastream.api.function.TwoInputNonBroadcastStreamProcessFunction; +import org.apache.flink.datastream.api.function.TwoOutputStreamProcessFunction; + +/** + * This interface represents a kind of partitioned data stream. For this stream, each parallelism is + * a partition, and the partition to which the data belongs is random. + */ +@Experimental +public interface NonKeyedPartitionStream<T> extends DataStream { + /** + * Apply an operation to this {@link NonKeyedPartitionStream}. + * + * @param processFunction to perform operation. + * @return new stream with this operation. + */ + <OUT> NonKeyedPartitionStream<OUT> process( + OneInputStreamProcessFunction<T, OUT> processFunction); + + /** + * Apply a two output operation to this {@link NonKeyedPartitionStream}. + * + * @param processFunction to perform two output operation. + * @return new stream with this operation. + */ + <OUT1, OUT2> TwoNonKeyedPartitionStreams<OUT1, OUT2> process( + TwoOutputStreamProcessFunction<T, OUT1, OUT2> processFunction); + + /** + * Apply to a two input operation on this and other {@link NonKeyedPartitionStream}. + * + * @param other {@link NonKeyedPartitionStream} to perform operation with two input. + * @param processFunction to perform operation. + * @return new stream with this operation. + */ + <T_OTHER, OUT> NonKeyedPartitionStream<OUT> connectAndProcess( + NonKeyedPartitionStream<T_OTHER> other, + TwoInputNonBroadcastStreamProcessFunction<T, T_OTHER, OUT> processFunction); + + /** + * Apply a two input operation to this and other {@link BroadcastStream}. + * + * @param processFunction to perform operation. + * @return new stream with this operation. + */ + <T_OTHER, OUT> NonKeyedPartitionStream<OUT> connectAndProcess( + BroadcastStream<T_OTHER> other, + TwoInputBroadcastStreamProcessFunction<T, T_OTHER, OUT> processFunction); + + /** + * Coalesce this stream to a {@link GlobalStream}. + * + * @return the coalesced global stream. + */ + GlobalStream<T> global(); + + /** + * Transform this stream to a {@link KeyedPartitionStream}. + * + * @param keySelector to decide how to map data to partition. + * @return the transformed stream partitioned by key. + */ + <K> KeyedPartitionStream<K, T> keyBy(KeySelector<T, K> keySelector); + + /** + * Transform this stream to a new {@link NonKeyedPartitionStream}, data will be shuffled between + * these two streams. + * + * @return the transformed stream after shuffle. + */ + NonKeyedPartitionStream<T> shuffle(); + + /** + * Transform this stream to a new {@link BroadcastStream}. + * + * @return the transformed {@link BroadcastStream}. + */ + BroadcastStream<T> broadcast(); + + // TODO add toSink method. + + /** + * This interface represents a combination of two {@link NonKeyedPartitionStream}. It will be + * used as the return value of operation with two output. + */ + @Experimental + interface TwoNonKeyedPartitionStreams<T1, T2> { + /** Get the first stream. */ + NonKeyedPartitionStream<T1> getFirst(); + + /** Get the second stream. */ + NonKeyedPartitionStream<T2> getSecond(); + } +} diff --git a/pom.xml b/pom.xml index aa8eaa4bbc6..e6d167129e4 100644 --- a/pom.xml +++ b/pom.xml @@ -2363,6 +2363,7 @@ under the License. <exclude>@org.apache.flink.annotation.Internal</exclude> <!-- MARKER: start exclusions; these will be wiped by tools/releasing/update_japicmp_configuration.sh --> <exclude>org.apache.flink.api.common.functions.Function</exclude> + <exclude>org.apache.flink.api.java.functions.KeySelector</exclude> <!-- MARKER: end exclusions --> </excludes> <accessModifier>public</accessModifier>
