nilmadhab commented on code in PR #26284: URL: https://github.com/apache/flink/pull/26284#discussion_r2015752289
########## flink-examples/flink-examples-streaming/src/main/java/org/apache/flink/streaming/examples/dsv2/wordcount/WordCountUsingLambda.java: ########## @@ -0,0 +1,265 @@ +/* + * 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.streaming.examples.dsv2.wordcount; + +import org.apache.flink.api.common.RuntimeExecutionMode; +import org.apache.flink.api.common.serialization.SimpleStringEncoder; +import org.apache.flink.api.common.state.StateDeclaration; +import org.apache.flink.api.common.state.StateDeclarations; +import org.apache.flink.api.common.state.ValueStateDeclaration; +import org.apache.flink.api.common.typeinfo.TypeDescriptors; +import org.apache.flink.api.common.typeinfo.TypeHint; +import org.apache.flink.api.common.typeinfo.TypeInformation; +import org.apache.flink.api.connector.dsv2.DataStreamV2SourceUtils; +import org.apache.flink.api.connector.dsv2.WrappedSink; +import org.apache.flink.api.connector.dsv2.WrappedSource; +import org.apache.flink.api.java.tuple.Tuple2; +import org.apache.flink.configuration.MemorySize; +import org.apache.flink.connector.file.sink.FileSink; +import org.apache.flink.connector.file.src.FileSource; +import org.apache.flink.connector.file.src.reader.TextLineInputFormat; +import org.apache.flink.core.fs.Path; +import org.apache.flink.datastream.api.ExecutionEnvironment; +import org.apache.flink.datastream.api.common.Collector; +import org.apache.flink.datastream.api.context.PartitionedContext; +import org.apache.flink.datastream.api.function.OneInputStreamProcessFunction; +import org.apache.flink.datastream.api.stream.KeyedPartitionStream; +import org.apache.flink.datastream.api.stream.NonKeyedPartitionStream; +import org.apache.flink.streaming.api.functions.sink.PrintSink; +import org.apache.flink.streaming.api.functions.sink.filesystem.rollingpolicies.DefaultRollingPolicy; +import org.apache.flink.streaming.examples.wordcount.util.WordCountData; +import org.apache.flink.util.ParameterTool; +import org.apache.flink.util.TimeUtils; + +import java.time.Duration; +import java.util.Arrays; +import java.util.Set; + +/** + * Implements the "WordCount" program by DataStream API V2 that computes a simple word occurrence + * histogram over text files. The job will currently be executed in streaming mode, and will support + * batch mode execution in the future. + * + * <p>The input is a [list of] plain text file[s] with lines separated by a newline character. + * + * <p>Usage: + * + * <ul> + * <li><code>--input <path></code>A list of input files and / or directories to read. If no + * input is provided, the program is run with default data from {@link WordCountData}. + * <li><code>--discovery-interval <duration></code>Turns the file reader into a continuous + * source that will monitor the provided input directories every interval and read any new + * files. + * <li><code>--output <path></code>The output directory where the Job will write the + * results. If no output path is provided, the Job will print the results to <code>stdout + * </code>. + * </ul> + * + * <p>This example shows how to: + * + * <ul> + * <li>Write a simple Flink program by DataStream API V2 + * <li>Use tuple data types + * <li>Write and use a user-defined process function + * </ul> + * + * <p>Please note that if you intend to run this example in an IDE, you must first add the following + * VM options: "--add-opens=java.base/java.util=ALL-UNNAMED". This is necessary because the module + * system in JDK 17+ restricts some reflection operations. + * + * <p>Please note that the DataStream API V2 is a new set of APIs, to gradually replace the original + * DataStream API. It is currently in the experimental stage and is not fully available for + * production. + */ +public class WordCountUsingLambda { Review Comment: yes, I was thinking the same ! ########## flink-datastream-api/src/main/java/org/apache/flink/datastream/api/stream/NonKeyedPartitionStream.java: ########## @@ -41,6 +42,20 @@ public interface NonKeyedPartitionStream<T> extends DataStream { <OUT> ProcessConfigurableAndNonKeyedPartitionStream<OUT> process( OneInputStreamProcessFunction<T, OUT> processFunction); + /** + * Adds a type information hint about the return type of this operator. This method can be used + * in cases where Flink cannot determine automatically what the produced type of a function is. + * That can be the case if the function uses generic type variables in the return type that + * cannot be inferred from the input type. + * + * <p>In most cases, the methods {@link #returns(Class)} and {@link #returns(TypeHint)} are + * preferable. + * + * @param typeInfo type information as a return type hint + * @return This operator with a given return type hint. + */ + NonKeyedPartitionStream<T> returns(TypeInformation<T> typeInfo); Review Comment: okay I will add it, -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
