You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by gu...@apache.org on 2025/02/18 01:32:23 UTC
(flink) 03/03: [FLINK-37274][datastream] Add examples for DataStream V2
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 3e3d556e0bb670c9027b6cf410604b4ac8bc2e9e
Author: Xu Huang <hu...@gmail.com>
AuthorDate: Fri Jan 24 10:29:39 2025 +0800
[FLINK-37274][datastream] Add examples for DataStream V2
---
.../impl/context/DefaultStateManager.java | 49 +-
flink-examples/flink-examples-streaming/pom.xml | 10 +
.../examples/dsv2/eventtime/CountNewsClicks.java | 351 +++++++++++++++
.../flink/streaming/examples/dsv2/join/Join.java | 213 +++++++++
.../examples/dsv2/watermark/CountSales.java | 277 ++++++++++++
.../dsv2/windowing/CountProductSalesWindowing.java | 200 +++++++++
.../examples/dsv2/wordcount/WordCount.java | 274 +++++++++++
.../datas/dsv2/eventtime/CountNewsClicksEvents.csv | 500 +++++++++++++++++++++
.../main/resources/datas/dsv2/join/JoinGrades.csv | 179 ++++++++
.../resources/datas/dsv2/join/JoinSalaries.csv | 179 ++++++++
.../windowing/CountProductSalesWindowingOrders.csv | 500 +++++++++++++++++++++
.../test/examples/DSv2ExamplesITCase.java | 104 +++++
.../datas/dsv2/eventtime/CountNewsClicksResult.csv | 20 +
.../test/resources/datas/dsv2/join/JoinResult.csv | 179 ++++++++
.../windowing/CountProductSalesWindowingResult.csv | 115 +++++
pom.xml | 2 +
16 files changed, 3118 insertions(+), 34 deletions(-)
diff --git a/flink-datastream/src/main/java/org/apache/flink/datastream/impl/context/DefaultStateManager.java b/flink-datastream/src/main/java/org/apache/flink/datastream/impl/context/DefaultStateManager.java
index 43949b59477..43222859ae4 100644
--- a/flink-datastream/src/main/java/org/apache/flink/datastream/impl/context/DefaultStateManager.java
+++ b/flink-datastream/src/main/java/org/apache/flink/datastream/impl/context/DefaultStateManager.java
@@ -92,12 +92,7 @@ public class DefaultStateManager implements StateManager {
@Override
public <T> ValueState<T> getState(ValueStateDeclaration<T> stateDeclaration) throws Exception {
Optional<ValueState<T>> stateOptional = getStateOptional(stateDeclaration);
- if (stateOptional.isEmpty()) {
- throw new RuntimeException(
- "Failed to access the State. You may need to declare it first and ensure that "
- + "the context can access the state. For example, retrieving a "
- + "keyed state is not allowed in a non-keyed context.");
- }
+ checkStateIsAvailable(stateOptional);
return stateOptional.get();
}
@@ -128,12 +123,7 @@ public class DefaultStateManager implements StateManager {
@Override
public <T> ListState<T> getState(ListStateDeclaration<T> stateDeclaration) throws Exception {
Optional<ListState<T>> stateOptional = getStateOptional(stateDeclaration);
- if (stateOptional.isEmpty()) {
- throw new RuntimeException(
- "Failed to access the State. You may need to declare it first and ensure that "
- + "the context can access the state. For example, retrieving a "
- + "keyed state is not allowed in a non-keyed context.");
- }
+ checkStateIsAvailable(stateOptional);
return stateOptional.get();
}
@@ -154,12 +144,7 @@ public class DefaultStateManager implements StateManager {
public <K, V> MapState<K, V> getState(MapStateDeclaration<K, V> stateDeclaration)
throws Exception {
Optional<MapState<K, V>> stateOptional = getStateOptional(stateDeclaration);
- if (stateOptional.isEmpty()) {
- throw new RuntimeException(
- "Failed to access the State. You may need to declare it first and ensure that "
- + "the context can access the state. For example, retrieving a "
- + "keyed state is not allowed in a non-keyed context.");
- }
+ checkStateIsAvailable(stateOptional);
return stateOptional.get();
}
@@ -179,12 +164,7 @@ public class DefaultStateManager implements StateManager {
public <T> ReducingState<T> getState(ReducingStateDeclaration<T> stateDeclaration)
throws Exception {
Optional<ReducingState<T>> stateOptional = getStateOptional(stateDeclaration);
- if (stateOptional.isEmpty()) {
- throw new RuntimeException(
- "Failed to access the State. You may need to declare it first and ensure that "
- + "the context can access the state. For example, retrieving a "
- + "keyed state is not allowed in a non-keyed context.");
- }
+ checkStateIsAvailable(stateOptional);
return stateOptional.get();
}
@@ -204,12 +184,7 @@ public class DefaultStateManager implements StateManager {
public <IN, ACC, OUT> AggregatingState<IN, OUT> getState(
AggregatingStateDeclaration<IN, ACC, OUT> stateDeclaration) throws Exception {
Optional<AggregatingState<IN, OUT>> stateOptional = getStateOptional(stateDeclaration);
- if (stateOptional.isEmpty()) {
- throw new RuntimeException(
- "Failed to access the State. You may need to declare it first and ensure that "
- + "the context can access the state. For example, retrieving a "
- + "keyed state is not allowed in a non-keyed context.");
- }
+ checkStateIsAvailable(stateOptional);
return stateOptional.get();
}
@@ -230,13 +205,19 @@ public class DefaultStateManager implements StateManager {
public <K, V> BroadcastState<K, V> getState(BroadcastStateDeclaration<K, V> stateDeclaration)
throws Exception {
Optional<BroadcastState<K, V>> stateOptional = getStateOptional(stateDeclaration);
+ checkStateIsAvailable(stateOptional);
+ return stateOptional.get();
+ }
+
+ private void checkStateIsAvailable(Optional<?> stateOptional) {
if (stateOptional.isEmpty()) {
- throw new RuntimeException(
+ throw new IllegalStateException(
"Failed to access the State. You may need to declare it first and ensure that "
- + "the context can access the state. For example, retrieving a "
- + "keyed state is not allowed in a non-keyed context.");
+ + "the context can access the State. For more information, please refer"
+ + " to 'https://nightlies.apache.org/flink/flink-docs-master/docs"
+ + "/dev/datastream-v2/context_and_state_processing"
+ + "/#the-legitimacy-of-state-declaration-and-access'.");
}
- return stateOptional.get();
}
/**
diff --git a/flink-examples/flink-examples-streaming/pom.xml b/flink-examples/flink-examples-streaming/pom.xml
index c2c99904425..4310b233654 100644
--- a/flink-examples/flink-examples-streaming/pom.xml
+++ b/flink-examples/flink-examples-streaming/pom.xml
@@ -36,6 +36,10 @@ under the License.
<!-- Allow users to pass custom jcuda versions -->
<properties>
<jcuda.version>10.0.0</jcuda.version>
+ <surefire.module.config><!--
+ DataStream V2 API examples
+ -->--add-opens=java.base/java.util=ALL-UNNAMED
+ </surefire.module.config>
</properties>
<dependencies>
@@ -66,6 +70,12 @@ under the License.
<version>3.0.0-1.17</version>
</dependency>
+ <dependency>
+ <groupId>org.apache.flink</groupId>
+ <artifactId>flink-csv</artifactId>
+ <version>${project.version}</version>
+ </dependency>
+
<dependency>
<groupId>org.apache.flink</groupId>
<artifactId>flink-connector-datagen</artifactId>
diff --git a/flink-examples/flink-examples-streaming/src/main/java/org/apache/flink/streaming/examples/dsv2/eventtime/CountNewsClicks.java b/flink-examples/flink-examples-streaming/src/main/java/org/apache/flink/streaming/examples/dsv2/eventtime/CountNewsClicks.java
new file mode 100644
index 00000000000..567179a0a09
--- /dev/null
+++ b/flink-examples/flink-examples-streaming/src/main/java/org/apache/flink/streaming/examples/dsv2/eventtime/CountNewsClicks.java
@@ -0,0 +1,351 @@
+/*
+ * 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.eventtime;
+
+import org.apache.flink.api.common.serialization.SimpleStringEncoder;
+import org.apache.flink.api.common.state.ListStateDeclaration;
+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.state.v2.ValueState;
+import org.apache.flink.api.common.typeinfo.TypeDescriptors;
+import org.apache.flink.api.connector.dsv2.WrappedSink;
+import org.apache.flink.api.connector.dsv2.WrappedSource;
+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.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.context.StateManager;
+import org.apache.flink.datastream.api.extension.eventtime.EventTimeExtension;
+import org.apache.flink.datastream.api.extension.eventtime.function.OneInputEventTimeStreamProcessFunction;
+import org.apache.flink.datastream.api.extension.eventtime.timer.EventTimeManager;
+import org.apache.flink.datastream.api.stream.NonKeyedPartitionStream;
+import org.apache.flink.formats.csv.CsvReaderFormat;
+import org.apache.flink.streaming.api.functions.sink.PrintSink;
+import org.apache.flink.streaming.api.functions.sink.filesystem.rollingpolicies.DefaultRollingPolicy;
+import org.apache.flink.util.ParameterTool;
+
+import java.time.Duration;
+import java.util.Objects;
+import java.util.Set;
+
+/**
+ * This example illustrates how to count the number of clicks on each news at 1 hour after news
+ * publication.
+ *
+ * <p>The input consists of a series of {@link NewsEvent}s, which fall into two categories: news
+ * publish and news click. Each {@link NewsEvent} contains three components: the event type, the
+ * news ID and the timestamp. Notably, there is only one event of type {@link NewsEventType#PUBLISH}
+ * for each news.
+ *
+ * <p>Usage:
+ *
+ * <ul>
+ * <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>Usage of Event Time extension in DataStream API V2
+ * </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 CountNewsClicks {
+
+ /**
+ * The type of {@link NewsEvent}, note that only one event of type {@link NewsEventType#PUBLISH}
+ * for each news.
+ */
+ public enum NewsEventType {
+ PUBLISH,
+ CLICK
+ }
+
+ /**
+ * The {@link NewsEvent} represents an event on news, containing the event type, news id and the
+ * timestamp.
+ */
+ public static class NewsEvent {
+ public long newsId;
+ public long timestamp;
+ public NewsEventType type;
+ }
+
+ /**
+ * The {@link NewsClicks} represents the number of clicks on news within one hour following its
+ * publication.
+ */
+ public static class NewsClicks {
+ public long newsId;
+ public long clicks;
+
+ public NewsClicks(long newsId, long clicks) {
+ this.newsId = newsId;
+ this.clicks = clicks;
+ }
+
+ @Override
+ public String toString() {
+ return String.format("%d,%d", this.newsId, this.clicks);
+ }
+ }
+
+ public static void main(String[] args) throws Exception {
+ // parse the parameters
+ final ParameterTool params = ParameterTool.fromArgs(args);
+ final boolean fileOutput = params.has("output");
+
+ // obtain execution environment
+ ExecutionEnvironment env = ExecutionEnvironment.getInstance();
+
+ // the input consists of a series of {code NewsEvent}s, which include two types: news
+ // publish event and news click event.
+ // We provide a CSV file to represent the sample input data, which can be found in the
+ // resources directory.
+ String inputFilePath =
+ Objects.requireNonNull(
+ CountNewsClicks.class
+ .getClassLoader()
+ .getResource(
+ "datas/dsv2/eventtime/CountNewsClicksEvents.csv"))
+ .getPath();
+ // create a FileSource with CSV format to read the input data
+ CsvReaderFormat<NewsEvent> csvFormat = CsvReaderFormat.forPojo(NewsEvent.class);
+ FileSource<NewsEvent> fileSource =
+ FileSource.forRecordStreamFormat(csvFormat, new Path(inputFilePath)).build();
+ NonKeyedPartitionStream<NewsEvent> source =
+ env.fromSource(new WrappedSource<>(fileSource), "news source");
+
+ NonKeyedPartitionStream<NewsClicks> clicksStream =
+ // extract event time and generate the event time watermark
+ source.process(
+ // the timestamp field of the input is considered to be the
+ // event time
+ EventTimeExtension.<NewsEvent>newWatermarkGeneratorBuilder(
+ event -> event.timestamp)
+ // generate event time watermarks every 200ms
+ .periodicWatermark(Duration.ofMillis(200))
+ // if the input is idle for more than 30 seconds, it
+ // is ignored during the event time watermark
+ // combination process
+ .withIdleness(Duration.ofSeconds(30))
+ // set the maximum out-of-order time of the event to
+ // 30 seconds, meaning that if an event is received
+ // at 12:00:00, then no further events should be
+ // received earlier than 11:59:30
+ .withMaxOutOfOrderTime(Duration.ofSeconds(10))
+ // build the event time watermark generator as
+ // ProcessFunction
+ .buildAsProcessFunction())
+ // key by the news id
+ .keyBy(event -> event.newsId)
+ // count the click number of each news
+ .process(
+ EventTimeExtension.wrapProcessFunction(
+ new CountNewsClicksProcessFunction()));
+
+ if (fileOutput) {
+ // write results to file
+ clicksStream
+ .toSink(
+ new WrappedSink<>(
+ FileSink.<NewsClicks>forRowFormat(
+ new Path(params.get("output")),
+ new SimpleStringEncoder<>())
+ .withRollingPolicy(
+ DefaultRollingPolicy.builder()
+ .withMaxPartSize(
+ MemorySize.ofMebiBytes(1))
+ .withRolloverInterval(
+ Duration.ofSeconds(10))
+ .build())
+ .build()))
+ .withName("output");
+ } else {
+ // Print the results to the STDOUT.
+ clicksStream.toSink(new WrappedSink<>(new PrintSink<>())).withName("print-sink");
+ }
+
+ env.execute("CountNewsClicks");
+ }
+
+ /**
+ * This process function will consume {@link NewsEvent} and count the number of clicks within 1
+ * hour of the news publication and send the results {@link NewsClicks} to the output.
+ *
+ * <p>To achieve the goal, we will register an event timer for each news, which will be
+ * triggered at the time of the news's release time + 1 hour, and record a click count of each
+ * news. In the timer callback {@code onEventTimer}, we will output the number of clicks.
+ *
+ * <p>To handle the potential disorder between news publication event and click events, we will
+ * use a ListState to store the timestamps of click events that occur prior to their
+ * corresponding publication events.
+ *
+ * <p>To handle the potential missing news publication event, we will register an event timer
+ * for the click event and set the timer to the timestamp of the click event plus one hour.
+ */
+ public static class CountNewsClicksProcessFunction
+ implements OneInputEventTimeStreamProcessFunction<NewsEvent, NewsClicks> {
+
+ private static final long ONE_HOUR_IN_MS = Duration.ofHours(1).toMillis();
+
+ private EventTimeManager eventTimeManager;
+
+ // uses a ValueState to store the publishing time of each news
+ private final ValueStateDeclaration<Long> publishTimeStateDeclaration =
+ StateDeclarations.valueState("publish_time", TypeDescriptors.LONG);
+
+ // uses a ValueState to store the click count of each news
+ private final ValueStateDeclaration<Long> clickCountStateDeclaration =
+ StateDeclarations.valueState("click_count", TypeDescriptors.LONG);
+
+ // uses a ListState to store the timestamp of pending clicks for each news.
+ // If a click {@code NewsEvent} occurs before the corresponding publication {@link
+ // NewsEvent}, we cannot process this click event because we do not know the publication
+ // time of the news.
+ // Therefore, we store the timestamp of this pending {@code NewsEvent} click in a list
+ // state.
+ // Once the publication {@link NewsEvent} arrives, we can proceed to process the pending
+ // clicks.
+ private final ListStateDeclaration<Long> pendingClicksStateDeclaration =
+ StateDeclarations.listState("pending_clicks", TypeDescriptors.LONG);
+
+ @Override
+ public void initEventTimeProcessFunction(EventTimeManager eventTimeManager) {
+ this.eventTimeManager = eventTimeManager;
+ }
+
+ @Override
+ public Set<StateDeclaration> usesStates() {
+ return Set.of(
+ publishTimeStateDeclaration,
+ clickCountStateDeclaration,
+ pendingClicksStateDeclaration);
+ }
+
+ @Override
+ public void processRecord(
+ NewsEvent record, Collector<NewsClicks> output, PartitionedContext<NewsClicks> ctx)
+ throws Exception {
+
+ if (record.type == NewsEventType.PUBLISH) {
+ // for the news publication event
+ long publishTime = record.timestamp;
+ if (eventTimeManager.currentTime() > publishTime + ONE_HOUR_IN_MS) {
+ // if the current event time is more than one hour after the publication time,
+ // it indicates that the publication event is late; therefore, this news should
+ // be ignored.
+ return;
+ }
+
+ // 1. record the publishing time
+ ctx.getStateManager().getState(publishTimeStateDeclaration).update(publishTime);
+
+ // 2. process pending clicks
+ Iterable<Long> pendingClicks =
+ ctx.getStateManager().getState(pendingClicksStateDeclaration).get();
+ if (pendingClicks != null) {
+ long addition = 0;
+ for (Long clickTime : pendingClicks) {
+ if (clickTime <= publishTime + ONE_HOUR_IN_MS) {
+ addition++;
+ }
+ }
+ addClickCount(ctx.getStateManager(), addition);
+ }
+
+ // 3. register timer with the publishing time + 1 hour
+ eventTimeManager.registerTimer(publishTime + ONE_HOUR_IN_MS);
+ } else {
+ // for the news click event
+ Long publishTime =
+ ctx.getStateManager().getState(publishTimeStateDeclaration).value();
+ if (publishTime == null) {
+ // If the news publication event has not yet arrived
+ // 1. record the click timestamp in the pending clicks list.
+ ctx.getStateManager()
+ .getState(pendingClicksStateDeclaration)
+ .add(record.timestamp);
+ // 2. to avoid the news not being processed for a long time when there is no
+ // publish event, register timer with the click timestamp + 1 hour.
+ eventTimeManager.registerTimer(record.timestamp + ONE_HOUR_IN_MS);
+ } else if (record.timestamp <= publishTime + ONE_HOUR_IN_MS) {
+ // If the click event occurs within one hour of the news publication, update the
+ // click count.
+ addClickCount(ctx.getStateManager(), 1);
+ }
+ }
+ }
+
+ @Override
+ public void onEventTimer(
+ long timestamp, Collector<NewsClicks> output, PartitionedContext<NewsClicks> ctx)
+ throws Exception {
+
+ Long publishTime = ctx.getStateManager().getState(publishTimeStateDeclaration).value();
+
+ if (publishTime == null) {
+ // If the publishing time is null, it indicates that there is no publishing event
+ // for
+ // this news, or the news has already been processed; therefore, it should be
+ // ignored.
+ clearAllState(ctx.getStateManager());
+ return;
+ }
+
+ // get the news that the current event timer belongs to
+ long newsId = ctx.getStateManager().getCurrentKey();
+ // get the click count of the news
+ Long clickCount = ctx.getStateManager().getState(clickCountStateDeclaration).value();
+ clickCount = clickCount == null ? 0 : clickCount;
+
+ // send the result to output
+ output.collect(new NewsClicks(newsId, clickCount));
+
+ // clear the state and mark the news as processed
+ clearAllState(ctx.getStateManager());
+ }
+
+ private void addClickCount(StateManager stateManager, long addition) throws Exception {
+ ValueState<Long> clickCountState = stateManager.getState(clickCountStateDeclaration);
+ Long previousCount = clickCountState.value();
+ long newlyCount = previousCount == null ? addition : previousCount + addition;
+ clickCountState.update(newlyCount);
+ }
+
+ private void clearAllState(StateManager stateManager) throws Exception {
+ stateManager.getState(publishTimeStateDeclaration).clear();
+ stateManager.getState(clickCountStateDeclaration).clear();
+ stateManager.getState(pendingClicksStateDeclaration).clear();
+ }
+ }
+}
diff --git a/flink-examples/flink-examples-streaming/src/main/java/org/apache/flink/streaming/examples/dsv2/join/Join.java b/flink-examples/flink-examples-streaming/src/main/java/org/apache/flink/streaming/examples/dsv2/join/Join.java
new file mode 100644
index 00000000000..1a2003df75d
--- /dev/null
+++ b/flink-examples/flink-examples-streaming/src/main/java/org/apache/flink/streaming/examples/dsv2/join/Join.java
@@ -0,0 +1,213 @@
+/*
+ * 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.join;
+
+import org.apache.flink.api.common.serialization.SimpleStringEncoder;
+import org.apache.flink.api.connector.dsv2.WrappedSink;
+import org.apache.flink.api.connector.dsv2.WrappedSource;
+import org.apache.flink.api.java.functions.KeySelector;
+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.core.fs.Path;
+import org.apache.flink.datastream.api.ExecutionEnvironment;
+import org.apache.flink.datastream.api.builtin.BuiltinFuncs;
+import org.apache.flink.datastream.api.common.Collector;
+import org.apache.flink.datastream.api.context.RuntimeContext;
+import org.apache.flink.datastream.api.extension.join.JoinFunction;
+import org.apache.flink.datastream.api.stream.NonKeyedPartitionStream;
+import org.apache.flink.formats.csv.CsvReaderFormat;
+import org.apache.flink.streaming.api.functions.sink.PrintSink;
+import org.apache.flink.streaming.api.functions.sink.filesystem.rollingpolicies.DefaultRollingPolicy;
+import org.apache.flink.util.ParameterTool;
+
+import java.time.Duration;
+import java.util.Objects;
+
+/**
+ * Example illustrating a join between two data streams. The example works on two input streams with
+ * pairs (name, grade) and (name, salary) respectively. It joins the streams based on "name".
+ *
+ * <p>The example uses a built-in sample data as inputs, which can be found in the resources
+ * directory.
+ *
+ * <p>Usage:
+ *
+ * <ul>
+ * <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>Usage of Join extension in DataStream API V2
+ * </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 Join {
+
+ /** POJO class for grades. */
+ public static class GradePojo {
+ public int grade;
+ public String name;
+ }
+
+ /** POJO class for salaries. */
+ public static class SalaryPojo {
+ public String name;
+ public long salary;
+ }
+
+ /** POJO class for grade and salary. */
+ public static class GradeAndSalaryPojo {
+ public String name;
+ public int grade;
+ public long salary;
+
+ public GradeAndSalaryPojo(String name, int grade, long salary) {
+ this.name = name;
+ this.grade = grade;
+ this.salary = salary;
+ }
+
+ public String toString() {
+ return String.format("%s,%d,%d", this.name, this.grade, this.salary);
+ }
+ }
+
+ public static void main(String[] args) throws Exception {
+ // parse the parameters
+ final ParameterTool params = ParameterTool.fromArgs(args);
+ final boolean fileOutput = params.has("output");
+
+ // obtain execution environment
+ ExecutionEnvironment env = ExecutionEnvironment.getInstance();
+
+ // create the data sources for both grades and salaries
+ // We provide CSV files to represent the sample input data, which can be found in the
+ // resources directory.
+ String gradesInputFilePath =
+ Objects.requireNonNull(
+ Join.class
+ .getClassLoader()
+ .getResource("datas/dsv2/join/JoinGrades.csv"))
+ .getPath();
+ // create a FileSource with CSV format to read the input data
+ FileSource<GradePojo> gradesFileSource =
+ FileSource.forRecordStreamFormat(
+ CsvReaderFormat.forPojo(GradePojo.class),
+ new Path(gradesInputFilePath))
+ .build();
+ NonKeyedPartitionStream<GradePojo> grades =
+ env.fromSource(new WrappedSource<>(gradesFileSource), "grade source");
+
+ String salariesInputFilePath =
+ Objects.requireNonNull(
+ Join.class
+ .getClassLoader()
+ .getResource("datas/dsv2/join/JoinSalaries.csv"))
+ .getPath();
+ FileSource<SalaryPojo> salariesFileSource =
+ FileSource.forRecordStreamFormat(
+ CsvReaderFormat.forPojo(SalaryPojo.class),
+ new Path(salariesInputFilePath))
+ .build();
+ NonKeyedPartitionStream<SalaryPojo> salaries =
+ env.fromSource(new WrappedSource<>(salariesFileSource), "salary source");
+
+ // joining two DataStreams
+ NonKeyedPartitionStream<GradeAndSalaryPojo> joinedStream =
+ BuiltinFuncs.join(
+ grades,
+ new GradeKeySelector(),
+ salaries,
+ new SalaryKeySelector(),
+ new JoinGradeAndSalaryFunction());
+
+ if (fileOutput) {
+ // write joined results to file
+ joinedStream
+ .toSink(
+ new WrappedSink<>(
+ FileSink.<GradeAndSalaryPojo>forRowFormat(
+ new Path(params.get("output")),
+ new SimpleStringEncoder<>())
+ .withRollingPolicy(
+ DefaultRollingPolicy.builder()
+ .withMaxPartSize(
+ MemorySize.ofMebiBytes(1))
+ .withRolloverInterval(
+ Duration.ofSeconds(10))
+ .build())
+ .build()))
+ .withName("output");
+ } else {
+ // Print the results to the STDOUT.
+ joinedStream.toSink(new WrappedSink<>(new PrintSink<>())).withName("print-sink");
+ }
+
+ // execute program
+ env.execute("Join Grade and Salary");
+ }
+
+ /** KeySelector of {@link GradePojo}. */
+ private static class GradeKeySelector implements KeySelector<GradePojo, String> {
+
+ @Override
+ public String getKey(GradePojo value) throws Exception {
+ return value.name;
+ }
+ }
+
+ /** KeySelector of {@link SalaryPojo}. */
+ private static class SalaryKeySelector implements KeySelector<SalaryPojo, String> {
+
+ @Override
+ public String getKey(SalaryPojo value) throws Exception {
+ return value.name;
+ }
+ }
+
+ /**
+ * Join the function of grade (name, grade) and salary (name, salary) to produce an output of
+ * (name, grade, salary).
+ */
+ private static class JoinGradeAndSalaryFunction
+ implements JoinFunction<GradePojo, SalaryPojo, GradeAndSalaryPojo> {
+
+ @Override
+ public void processRecord(
+ GradePojo leftRecord,
+ SalaryPojo rightRecord,
+ Collector<GradeAndSalaryPojo> output,
+ RuntimeContext ctx)
+ throws Exception {
+ output.collect(
+ new GradeAndSalaryPojo(leftRecord.name, leftRecord.grade, rightRecord.salary));
+ }
+ }
+}
diff --git a/flink-examples/flink-examples-streaming/src/main/java/org/apache/flink/streaming/examples/dsv2/watermark/CountSales.java b/flink-examples/flink-examples-streaming/src/main/java/org/apache/flink/streaming/examples/dsv2/watermark/CountSales.java
new file mode 100644
index 00000000000..2545fab2913
--- /dev/null
+++ b/flink-examples/flink-examples-streaming/src/main/java/org/apache/flink/streaming/examples/dsv2/watermark/CountSales.java
@@ -0,0 +1,277 @@
+/*
+ * 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.watermark;
+
+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.state.v2.ValueState;
+import org.apache.flink.api.common.typeinfo.TypeDescriptors;
+import org.apache.flink.api.common.typeinfo.TypeInformation;
+import org.apache.flink.api.common.watermark.LongWatermark;
+import org.apache.flink.api.common.watermark.LongWatermarkDeclaration;
+import org.apache.flink.api.common.watermark.Watermark;
+import org.apache.flink.api.common.watermark.WatermarkDeclaration;
+import org.apache.flink.api.common.watermark.WatermarkDeclarations;
+import org.apache.flink.api.common.watermark.WatermarkHandlingResult;
+import org.apache.flink.api.connector.dsv2.WrappedSink;
+import org.apache.flink.api.connector.dsv2.WrappedSource;
+import org.apache.flink.api.connector.source.SourceReaderContext;
+import org.apache.flink.connector.datagen.source.DataGeneratorSource;
+import org.apache.flink.connector.datagen.source.GeneratorFunction;
+import org.apache.flink.datastream.api.ExecutionEnvironment;
+import org.apache.flink.datastream.api.common.Collector;
+import org.apache.flink.datastream.api.context.NonPartitionedContext;
+import org.apache.flink.datastream.api.context.PartitionedContext;
+import org.apache.flink.datastream.api.function.ApplyPartitionFunction;
+import org.apache.flink.datastream.api.function.OneInputStreamProcessFunction;
+import org.apache.flink.datastream.api.stream.NonKeyedPartitionStream;
+import org.apache.flink.streaming.api.functions.sink.PrintSink;
+import org.apache.flink.streaming.examples.dsv2.eventtime.CountNewsClicks;
+import org.apache.flink.util.ParameterTool;
+
+import java.time.Duration;
+import java.util.Set;
+import java.util.concurrent.ThreadLocalRandom;
+
+/**
+ * This example shows how to count the cumulative sales of each product at the moment. In this
+ * example, we simulate the creation and propagation of event time watermarks through {@link
+ * Watermark}.
+ *
+ * <p>Please note that Flink provides users with an event time extension to support event time. You
+ * can refer to the {@link CountNewsClicks} example.
+ *
+ * <p>The example uses a {@link DataGeneratorSource} as input source and a custom {@link Watermark}
+ * to propagate the event time in the stream.
+ *
+ * <p>Usage:
+ *
+ * <ul>
+ * <li><code>--parallelism <path></code>The parallelism of the source. The default value is
+ * 5.
+ * </ul>
+ *
+ * <p>This example shows how to:
+ *
+ * <ul>
+ * <li>How to define and propagate custom event (Watermark) in DataStream API V2
+ * </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 CountSales {
+
+ /** Pojo class for Order. */
+ public static class Order {
+ public long productId;
+ public double totalPrice;
+ public long timestamp;
+
+ public Order(long productId, double price, long timestamp) {
+ this.productId = productId;
+ this.totalPrice = price;
+ this.timestamp = timestamp;
+ }
+ }
+
+ /** {@link CumulativeSales} represents the cumulative sales at a certain moment of a product. */
+ public static class CumulativeSales {
+ public long productId;
+ public long timestamp;
+ public double sales;
+
+ public CumulativeSales(long productId, long timestamp, double sales) {
+ this.productId = productId;
+ this.timestamp = timestamp;
+ this.sales = sales;
+ }
+
+ @Override
+ public String toString() {
+ return String.format("%d,%d,%.2f", this.productId, this.timestamp, this.sales);
+ }
+ }
+
+ /**
+ * Firstly, we define an event time watermark, which represents the time of currently processing
+ * event. Since the watermark needs to convey the timestamp, its data type is long. To determine
+ * the minimum event time across all watermarks, we utilize the combineFunctionMin() method to
+ * combine the watermarks. The default handling strategy is forward, meaning that the watermark
+ * will typically be advanced to downstream operators in most scenarios. Thus, we create a
+ * WatermarkDeclaration instance that can be used to declare and generate the watermark.
+ */
+ public static final LongWatermarkDeclaration EVENT_TIME_WATERMARK_DECLARATION =
+ WatermarkDeclarations.newBuilder("EVENT_TIME")
+ .typeLong()
+ .combineFunctionMin()
+ .combineWaitForAllChannels(true)
+ .defaultHandlingStrategyForward()
+ .build();
+
+ public static void main(String[] args) throws Exception {
+ // parse the parameters
+ final ParameterTool params = ParameterTool.fromArgs(args);
+ final int parallelism = params.getInt("parallelism", 5);
+
+ // obtain execution environment
+ ExecutionEnvironment env = ExecutionEnvironment.getInstance();
+
+ // Create the Order source, the source will declare and generate event time watermarks.
+ NonKeyedPartitionStream<Order> source =
+ env.fromSource(new WrappedSource<>(new OrderSource()), "order source")
+ .withParallelism(parallelism);
+
+ source
+ // key by product id
+ .keyBy(order -> order.productId)
+ .process(
+ // handle event time watermark in downstream
+ new CountSalesProcessFunction())
+ .toSink(new WrappedSink<>(new PrintSink<>()));
+
+ // execute program
+ env.execute("Count Sales");
+ }
+
+ /** Source of Orders. We will declare and generate the event time watermark in this source. */
+ private static class OrderSource extends DataGeneratorSource<Order> {
+
+ public OrderSource() {
+ super(new CustomGeneratorFunction(), 1_000_000L, TypeInformation.of(Order.class));
+ }
+
+ @Override
+ public Set<? extends WatermarkDeclaration> declareWatermarks() {
+ // Declare the event time watermark.
+ return Set.of(EVENT_TIME_WATERMARK_DECLARATION);
+ }
+ }
+
+ /**
+ * Generator function for the Order source. We will generate event time watermark every second
+ * within this generator function.
+ */
+ private static class CustomGeneratorFunction implements GeneratorFunction<Long, Order> {
+
+ private SourceReaderContext readerContext;
+
+ private long previousTimestamp = 0;
+
+ @Override
+ public void open(SourceReaderContext readerContext) throws Exception {
+ this.readerContext = readerContext;
+ }
+
+ @Override
+ public Order map(Long value) throws Exception {
+ long productId = ThreadLocalRandom.current().nextLong(1, 100);
+ double price = ThreadLocalRandom.current().nextDouble(1, 1000);
+ long timestamp = System.currentTimeMillis();
+ if (timestamp - previousTimestamp > Duration.ofSeconds(1).toMillis()) {
+ // Send event time watermark
+ readerContext.emitWatermark(
+ EVENT_TIME_WATERMARK_DECLARATION.newWatermark(timestamp));
+ previousTimestamp = timestamp;
+ }
+
+ Thread.sleep(100);
+ return new Order(productId, price, timestamp);
+ }
+ }
+
+ /**
+ * This process function calculates the cumulative sales for each product. It utilizes a
+ * ValueState to store the cumulative sales and updates this state whenever a new record is
+ * received. Additionally, it emits the cumulative sales for each product when an event-time
+ * watermark arrives.
+ */
+ private static class CountSalesProcessFunction
+ implements OneInputStreamProcessFunction<Order, CumulativeSales> {
+
+ // uses a ValueState to store the sales of each product
+ private final ValueStateDeclaration<Double> salesStateDeclaration =
+ StateDeclarations.valueState("sales", TypeDescriptors.DOUBLE);
+
+ @Override
+ public Set<StateDeclaration> usesStates() {
+ return Set.of(salesStateDeclaration);
+ }
+
+ @Override
+ public void processRecord(
+ Order record,
+ Collector<CumulativeSales> output,
+ PartitionedContext<CumulativeSales> ctx)
+ throws Exception {
+ // update the sales state
+ ValueState<Double> salesState = ctx.getStateManager().getState(salesStateDeclaration);
+ Double previousSales = salesState.value();
+ Double newlySales =
+ previousSales == null ? record.totalPrice : previousSales + record.totalPrice;
+ salesState.update(newlySales);
+ }
+
+ @Override
+ public WatermarkHandlingResult onWatermark(
+ Watermark watermark,
+ Collector<CumulativeSales> output,
+ NonPartitionedContext<CumulativeSales> ctx) {
+ if (watermark
+ .getIdentifier()
+ .equals(EVENT_TIME_WATERMARK_DECLARATION.getIdentifier())) {
+ // For our custom event time watermark, we can get current event time and output
+ // sales result
+ long currentEventTime = ((LongWatermark) watermark).getValue();
+ try {
+ ctx.applyToAllPartitions(
+ new ApplyPartitionFunction<CumulativeSales>() {
+
+ @Override
+ public void apply(
+ Collector<CumulativeSales> collector,
+ PartitionedContext<CumulativeSales> ctx)
+ throws Exception {
+ long productId = ctx.getStateManager().getCurrentKey();
+ Double sales =
+ ctx.getStateManager()
+ .getState(salesStateDeclaration)
+ .value();
+ sales = sales == null ? 0 : sales;
+ collector.collect(
+ new CumulativeSales(
+ productId, currentEventTime, sales));
+ }
+ });
+ } catch (Exception e) {
+ throw new RuntimeException(e);
+ }
+
+ return WatermarkHandlingResult.PEEK;
+ }
+
+ return WatermarkHandlingResult.PEEK;
+ }
+ }
+}
diff --git a/flink-examples/flink-examples-streaming/src/main/java/org/apache/flink/streaming/examples/dsv2/windowing/CountProductSalesWindowing.java b/flink-examples/flink-examples-streaming/src/main/java/org/apache/flink/streaming/examples/dsv2/windowing/CountProductSalesWindowing.java
new file mode 100644
index 00000000000..41c9aa924d7
--- /dev/null
+++ b/flink-examples/flink-examples-streaming/src/main/java/org/apache/flink/streaming/examples/dsv2/windowing/CountProductSalesWindowing.java
@@ -0,0 +1,200 @@
+/*
+ * 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.windowing;
+
+import org.apache.flink.api.common.serialization.SimpleStringEncoder;
+import org.apache.flink.api.connector.dsv2.WrappedSink;
+import org.apache.flink.api.connector.dsv2.WrappedSource;
+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.core.fs.Path;
+import org.apache.flink.datastream.api.ExecutionEnvironment;
+import org.apache.flink.datastream.api.builtin.BuiltinFuncs;
+import org.apache.flink.datastream.api.common.Collector;
+import org.apache.flink.datastream.api.context.PartitionedContext;
+import org.apache.flink.datastream.api.extension.eventtime.EventTimeExtension;
+import org.apache.flink.datastream.api.extension.window.context.OneInputWindowContext;
+import org.apache.flink.datastream.api.extension.window.function.OneInputWindowStreamProcessFunction;
+import org.apache.flink.datastream.api.extension.window.strategy.WindowStrategy;
+import org.apache.flink.datastream.api.stream.NonKeyedPartitionStream;
+import org.apache.flink.formats.csv.CsvReaderFormat;
+import org.apache.flink.streaming.api.functions.sink.PrintSink;
+import org.apache.flink.streaming.api.functions.sink.filesystem.rollingpolicies.DefaultRollingPolicy;
+import org.apache.flink.util.ParameterTool;
+
+import java.time.Duration;
+import java.util.Objects;
+
+/**
+ * Example illustrating how to use Window to count the sales of each product in each hour by
+ * DataStream API V2.
+ *
+ * <p>The input is a list of orders, each order is a consist of productId and orderTimestamp.
+ *
+ * <p>Usage:
+ *
+ * <ul>
+ * <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>Usage of Window extension in DataStream API V2
+ * </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 CountProductSalesWindowing {
+
+ /** Pojo class for order. */
+ public static class Order {
+ public long productId;
+ public long timestamp;
+ }
+
+ /** The {@link ProductSales} class represents the count of product sales within one hour. */
+ public static class ProductSales {
+ public long productId;
+ public long startTime;
+ public long salesQuantity;
+
+ public ProductSales(long productId, long startTime, long salesQuantity) {
+ this.productId = productId;
+ this.startTime = startTime;
+ this.salesQuantity = salesQuantity;
+ }
+
+ @Override
+ public String toString() {
+ return String.format("%d,%d,%d", this.productId, this.startTime, this.salesQuantity);
+ }
+ }
+
+ public static void main(String[] args) throws Exception {
+ // parse the parameters
+ final ParameterTool params = ParameterTool.fromArgs(args);
+ final boolean fileOutput = params.has("output");
+
+ // Get the execution environment instance. This is the main entrypoint
+ // to building a Flink application.
+ final ExecutionEnvironment env = ExecutionEnvironment.getInstance();
+
+ // create order source stream
+ // We provide CSV files to represent the sample input data, which can be found in the
+ // resources directory.
+ String inputFilePath =
+ Objects.requireNonNull(
+ CountProductSalesWindowing.class
+ .getClassLoader()
+ .getResource(
+ "datas/dsv2/windowing/CountProductSalesWindowingOrders.csv"))
+ .getPath();
+ // create a FileSource with CSV format to read the input data
+ FileSource<Order> fileSource =
+ FileSource.forRecordStreamFormat(
+ CsvReaderFormat.forPojo(Order.class), new Path(inputFilePath))
+ .build();
+ NonKeyedPartitionStream<Order> orders =
+ env.fromSource(new WrappedSource<>(fileSource), "order source");
+
+ // extract and propagate event time from order
+ NonKeyedPartitionStream<Order> orderStream =
+ orders.process(
+ EventTimeExtension.<Order>newWatermarkGeneratorBuilder(
+ order -> order.timestamp)
+ .periodicWatermark(Duration.ofMillis(200))
+ .buildAsProcessFunction());
+
+ NonKeyedPartitionStream<ProductSales> productSalesQuantityStream =
+ orderStream
+ // key by productId
+ .keyBy(order -> order.productId)
+ .process(
+ BuiltinFuncs.window(
+ // declare tumbling window with window size 1 hour
+ WindowStrategy.tumbling(
+ Duration.ofHours(1), WindowStrategy.EVENT_TIME),
+ // define window process function to calculate total sales
+ // quantity per product per window.
+ new CountSalesQuantity()));
+
+ if (fileOutput) {
+ productSalesQuantityStream
+ .toSink(
+ new WrappedSink<>(
+ FileSink.<ProductSales>forRowFormat(
+ new Path(params.get("output")),
+ new SimpleStringEncoder<>())
+ .withRollingPolicy(
+ DefaultRollingPolicy.builder()
+ .withMaxPartSize(
+ MemorySize.ofMebiBytes(1))
+ .withRolloverInterval(
+ Duration.ofSeconds(10))
+ .build())
+ .build()))
+ .withName("file-sink");
+ } else {
+ // Print the results to the STDOUT.
+ productSalesQuantityStream
+ .toSink(new WrappedSink<>(new PrintSink<>()))
+ .withName("print-sink");
+ }
+
+ env.execute("Count Product Sales Windowing");
+ }
+
+ /**
+ * Count sales quantity per product.
+ *
+ * <p>We will obtain all orders and calculate the total sales quantity for each product when
+ * window trigger.
+ */
+ public static class CountSalesQuantity
+ implements OneInputWindowStreamProcessFunction<Order, ProductSales> {
+
+ @Override
+ public void onTrigger(
+ Collector<ProductSales> output,
+ PartitionedContext<ProductSales> ctx,
+ OneInputWindowContext<Order> windowContext)
+ throws Exception {
+ // get current productId
+ long productId = ctx.getStateManager().getCurrentKey();
+ // calculate total sales quantity
+ long salesQuantity = 0;
+ for (Order ignored : windowContext.getAllRecords()) {
+ salesQuantity += 1;
+ }
+ // emit result
+ output.collect(
+ new ProductSales(productId, windowContext.getStartTime(), salesQuantity));
+ }
+ }
+}
diff --git a/flink-examples/flink-examples-streaming/src/main/java/org/apache/flink/streaming/examples/dsv2/wordcount/WordCount.java b/flink-examples/flink-examples-streaming/src/main/java/org/apache/flink/streaming/examples/dsv2/wordcount/WordCount.java
new file mode 100644
index 00000000000..1f99c1c8fd4
--- /dev/null
+++ b/flink-examples/flink-examples-streaming/src/main/java/org/apache/flink/streaming/examples/dsv2/wordcount/WordCount.java
@@ -0,0 +1,274 @@
+/*
+ * 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.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 WordCount {
+
+ // *************************************************************************
+ // PROGRAM
+ // *************************************************************************
+
+ public static void main(String[] args) throws Exception {
+ // parse the parameters
+ final ParameterTool params = ParameterTool.fromArgs(args);
+
+ // Get the execution environment instance. This is the main entrypoint
+ // to building a Flink application.
+ final ExecutionEnvironment env = ExecutionEnvironment.getInstance();
+
+ // Apache Flink’s unified approach to stream and batch processing means that a DataStream
+ // application executed over bounded input will produce the same final results regardless
+ // of the configured execution mode. It is important to note what final means here: a job
+ // executing in STREAMING mode might produce incremental updates (think upserts in
+ // a database) while in BATCH mode, it would only produce one final result at the end. The
+ // final result will be the same if interpreted correctly, but getting there can be
+ // different.
+ //
+ // The “classic” execution behavior of the DataStream API is called STREAMING execution
+ // mode. Applications should use streaming execution for unbounded jobs that require
+ // continuous incremental processing and are expected to stay online indefinitely.
+ //
+ // By enabling BATCH execution, we allow Flink to apply additional optimizations that we
+ // can only do when we know that our input is bounded. For example, different
+ // join/aggregation strategies can be used, in addition to a different shuffle
+ // implementation that allows more efficient task scheduling and failure recovery behavior.
+ //
+ // By setting the runtime mode to AUTOMATIC, Flink will choose BATCH if all sources
+ // are bounded and otherwise STREAMING.
+ //
+ // This job will currently be executed in streaming mode, and will support batch mode
+ // execution in the future.
+ // TODO Support batch mode execution after the completion of the FLINK-37290 issue.
+ env.setExecutionMode(RuntimeExecutionMode.STREAMING);
+
+ NonKeyedPartitionStream<String> text;
+
+ if (params.has("input")) {
+ // Create a new file source that will read files from a given set of directories.
+ // Each file will be processed as plain text and split based on newlines.
+ FileSource.FileSourceBuilder<String> builder =
+ FileSource.forRecordStreamFormat(
+ new TextLineInputFormat(), new Path(params.get("input")));
+
+ // If a discovery interval is provided, the source will
+ // continuously watch the given directories for new files.
+ if (params.has("discovery-interval")) {
+ Duration discoveryInterval =
+ TimeUtils.parseDuration(params.get("discovery-interval"));
+ builder.monitorContinuously(discoveryInterval);
+ }
+
+ text = env.fromSource(new WrappedSource<>(builder.build()), "file-input");
+ } else {
+ // Create a new from data source with default data {@code WordCountData}.
+ text =
+ env.fromSource(
+ DataStreamV2SourceUtils.fromData(Arrays.asList(WordCountData.WORDS)),
+ "in-memory-input");
+ }
+
+ KeyedPartitionStream<String, Tuple2<String, Integer>> keyedStream =
+ // The text lines read from the source are split into words
+ // using a user-defined process function. The tokenizer, implemented below,
+ // will output each word as a (2-tuple) containing (word, 1)
+ text.process(new Tokenizer())
+ .withName("tokenizer")
+ // keyBy groups tuples based on the first field, the word.
+ // Using a keyBy allows performing aggregations and other
+ // stateful transformations over data on a per-key basis.
+ // This is similar to a GROUP BY clause in a SQL query.
+ .keyBy(value -> value.f0);
+
+ // For each key, we perform a simple sum of the second field, the count by user-defined
+ // Counter process function. It will continuously output updates each time it sees
+ // a new instance of each word in the stream.
+ // Flink will provide more convenient built-in functions, such as aggregation, in the
+ // future. This will
+ // allow users to avoid having to write their own Counter process
+ // functions in this application.
+ NonKeyedPartitionStream<Tuple2<String, Integer>> counts =
+ keyedStream.process(new Counter());
+
+ if (params.has("output")) {
+ // Given an output directory, Flink will write the results to a file
+ // using a simple string encoding. In a production environment, this might
+ // be something more structured like CSV, Avro, JSON, or Parquet.
+ counts.toSink(
+ new WrappedSink<>(
+ FileSink.<Tuple2<String, Integer>>forRowFormat(
+ new Path(params.get("output")),
+ new SimpleStringEncoder<>())
+ .withRollingPolicy(
+ DefaultRollingPolicy.builder()
+ .withMaxPartSize(
+ MemorySize.ofMebiBytes(1))
+ .withRolloverInterval(
+ Duration.ofSeconds(10))
+ .build())
+ .build()))
+ .withName("file-sink");
+ } else {
+ // Print the results to the STDOUT.
+ counts.toSink(new WrappedSink<>(new PrintSink<>())).withName("print-sink");
+ }
+
+ // Apache Flink applications are composed lazily. Calling execute
+ // submits the Job and begins processing.
+ env.execute("WordCount");
+ }
+
+ // *************************************************************************
+ // USER PROCESS FUNCTIONS
+ // *************************************************************************
+
+ /**
+ * Implements the string tokenizer that splits sentences into words as a user-defined
+ * ProcessFunction. The process function takes a line (String) and splits it into multiple pairs
+ * in the form of "(word,1)" ({@code Tuple2<String, Integer>}).
+ */
+ public static final class Tokenizer
+ implements OneInputStreamProcessFunction<String, Tuple2<String, Integer>> {
+
+ @Override
+ public void processRecord(
+ String record,
+ Collector<Tuple2<String, Integer>> output,
+ PartitionedContext<Tuple2<String, Integer>> ctx)
+ throws Exception {
+ // normalize and split the line
+ String[] tokens = record.toLowerCase().split("\\W+");
+
+ // emit the pairs
+ for (String token : tokens) {
+ if (!token.isEmpty()) {
+ output.collect(new Tuple2<>(token, 1));
+ }
+ }
+ }
+ }
+
+ /**
+ * Implements a word counter as a user-defined ProcessFunction that counts received words in
+ * streaming mode. The function uses a ValueState to store the count of each word, it will
+ * update the count of word and output the result when receive a record "(word,1)".
+ *
+ * <p>Note that this is just an example of how to code a streaming job using the DataStream API
+ * V2. It currently involves some complexity. In the future, we will provide more user-friendly
+ * APIs and extensions to simplify the process.
+ */
+ public static final class Counter
+ implements OneInputStreamProcessFunction<
+ Tuple2<String, Integer>, Tuple2<String, Integer>> {
+
+ // uses a ValueState to store the count of each word
+ private final ValueStateDeclaration<Integer> countStateDeclaration =
+ StateDeclarations.valueState("count", TypeDescriptors.INT);
+
+ @Override
+ public Set<StateDeclaration> usesStates() {
+ // declare a ValueState to store the count of each word
+ return Set.of(countStateDeclaration);
+ }
+
+ @Override
+ public void processRecord(
+ Tuple2<String, Integer> record,
+ Collector<Tuple2<String, Integer>> output,
+ PartitionedContext<Tuple2<String, Integer>> ctx)
+ throws Exception {
+ // calculate the new count of the word
+ String word = record.f0;
+ Integer count = record.f1;
+ Integer previousCount = ctx.getStateManager().getState(countStateDeclaration).value();
+ Integer newlyCount = previousCount == null ? count : previousCount + count;
+
+ // update the count of the word
+ ctx.getStateManager().getState(countStateDeclaration).update(newlyCount);
+
+ // output the result
+ output.collect(Tuple2.of(word, newlyCount));
+ }
+ }
+}
diff --git a/flink-examples/flink-examples-streaming/src/main/resources/datas/dsv2/eventtime/CountNewsClicksEvents.csv b/flink-examples/flink-examples-streaming/src/main/resources/datas/dsv2/eventtime/CountNewsClicksEvents.csv
new file mode 100644
index 00000000000..7363db4bcb6
--- /dev/null
+++ b/flink-examples/flink-examples-streaming/src/main/resources/datas/dsv2/eventtime/CountNewsClicksEvents.csv
@@ -0,0 +1,500 @@
+9, 1738830723194, PUBLISH
+16, 1738829879466, PUBLISH
+5, 1738830949115, PUBLISH
+3, 1738832288598, PUBLISH
+10, 1738828444792, PUBLISH
+9, 1738829535104, CLICK
+14, 1738828135370, PUBLISH
+1, 1738828782145, PUBLISH
+7, 1738830493544, PUBLISH
+13, 1738831726293, PUBLISH
+16, 1738829011506, CLICK
+11, 1738831106311, PUBLISH
+10, 1738830908408, CLICK
+17, 1738830252678, PUBLISH
+6, 1738828987697, PUBLISH
+10, 1738832605946, CLICK
+17, 1738831531284, CLICK
+20, 1738830288108, PUBLISH
+6, 1738830872381, CLICK
+7, 1738828614208, CLICK
+16, 1738827731027, CLICK
+7, 1738829971876, CLICK
+9, 1738832726137, CLICK
+11, 1738828223034, CLICK
+6, 1738829376088, CLICK
+6, 1738829917284, CLICK
+10, 1738831966069, CLICK
+15, 1738828413701, PUBLISH
+12, 1738828976311, PUBLISH
+11, 1738830811460, CLICK
+5, 1738830326169, CLICK
+5, 1738830016848, CLICK
+14, 1738831073351, CLICK
+10, 1738828609776, CLICK
+1, 1738828653882, CLICK
+17, 1738827847711, CLICK
+14, 1738829802795, CLICK
+18, 1738830108628, PUBLISH
+3, 1738827627444, CLICK
+11, 1738830503224, CLICK
+20, 1738831859047, CLICK
+14, 1738832139163, CLICK
+12, 1738830695626, CLICK
+6, 1738829273629, CLICK
+7, 1738832232320, CLICK
+9, 1738830973688, CLICK
+12, 1738832100794, CLICK
+20, 1738827800696, CLICK
+20, 1738832076878, CLICK
+10, 1738829791478, CLICK
+16, 1738831796537, CLICK
+17, 1738831451967, CLICK
+18, 1738832453204, CLICK
+20, 1738831377371, CLICK
+17, 1738830847549, CLICK
+18, 1738832439962, CLICK
+9, 1738832323264, CLICK
+3, 1738831606977, CLICK
+16, 1738830373365, CLICK
+13, 1738832642448, CLICK
+14, 1738831196971, CLICK
+1, 1738831782173, CLICK
+7, 1738832752364, CLICK
+7, 1738829411017, CLICK
+3, 1738828032429, CLICK
+20, 1738829630342, CLICK
+7, 1738831275619, CLICK
+13, 1738829775398, CLICK
+12, 1738829990639, CLICK
+14, 1738828868288, CLICK
+7, 1738830732103, CLICK
+8, 1738828425930, PUBLISH
+1, 1738828265666, CLICK
+12, 1738827974957, CLICK
+16, 1738828105580, CLICK
+3, 1738828752993, CLICK
+10, 1738828839294, CLICK
+5, 1738828353527, CLICK
+11, 1738829874555, CLICK
+8, 1738828939579, CLICK
+10, 1738831254013, CLICK
+13, 1738827748595, CLICK
+7, 1738832520783, CLICK
+8, 1738829861630, CLICK
+20, 1738831423761, CLICK
+9, 1738828541625, CLICK
+9, 1738828615616, CLICK
+13, 1738828624449, CLICK
+16, 1738828470370, CLICK
+1, 1738829739335, CLICK
+15, 1738830890342, CLICK
+14, 1738831904851, CLICK
+4, 1738829549171, PUBLISH
+17, 1738832606626, CLICK
+11, 1738829086800, CLICK
+12, 1738829004727, CLICK
+17, 1738831966274, CLICK
+13, 1738831983879, CLICK
+8, 1738831463524, CLICK
+19, 1738828812783, PUBLISH
+16, 1738832588205, CLICK
+1, 1738829517968, CLICK
+1, 1738828546565, CLICK
+6, 1738828673019, CLICK
+6, 1738828526042, CLICK
+1, 1738828406778, CLICK
+19, 1738830202196, CLICK
+7, 1738830301933, CLICK
+16, 1738832228567, CLICK
+16, 1738828547638, CLICK
+14, 1738829443511, CLICK
+17, 1738831344801, CLICK
+20, 1738829727975, CLICK
+18, 1738830457085, CLICK
+11, 1738832147312, CLICK
+16, 1738828175958, CLICK
+9, 1738831519150, CLICK
+1, 1738831455136, CLICK
+6, 1738832134053, CLICK
+6, 1738832062884, CLICK
+9, 1738831147735, CLICK
+3, 1738830849557, CLICK
+5, 1738831630602, CLICK
+9, 1738831417663, CLICK
+17, 1738828025677, CLICK
+11, 1738829268113, CLICK
+11, 1738830984646, CLICK
+12, 1738827972798, CLICK
+16, 1738830379897, CLICK
+9, 1738830389833, CLICK
+11, 1738832572229, CLICK
+2, 1738831857190, PUBLISH
+18, 1738827829058, CLICK
+20, 1738831761094, CLICK
+8, 1738828774815, CLICK
+17, 1738832551817, CLICK
+14, 1738829027223, CLICK
+6, 1738832305514, CLICK
+11, 1738828795952, CLICK
+1, 1738832186252, CLICK
+4, 1738832375819, CLICK
+9, 1738831124440, CLICK
+10, 1738830349924, CLICK
+13, 1738831754328, CLICK
+10, 1738832472785, CLICK
+15, 1738829597954, CLICK
+17, 1738830644441, CLICK
+13, 1738831342853, CLICK
+10, 1738832840939, CLICK
+16, 1738828536320, CLICK
+13, 1738828184808, CLICK
+9, 1738832690270, CLICK
+18, 1738832042347, CLICK
+7, 1738830289288, CLICK
+11, 1738829535267, CLICK
+2, 1738832685567, CLICK
+20, 1738828498710, CLICK
+14, 1738832392317, CLICK
+11, 1738832111139, CLICK
+11, 1738830164678, CLICK
+3, 1738832090451, CLICK
+14, 1738830404529, CLICK
+16, 1738828622575, CLICK
+18, 1738829119430, CLICK
+3, 1738829388500, CLICK
+10, 1738829630337, CLICK
+5, 1738828374887, CLICK
+3, 1738832939980, CLICK
+5, 1738830257751, CLICK
+9, 1738832950513, CLICK
+19, 1738830142852, CLICK
+19, 1738828718616, CLICK
+16, 1738831982315, CLICK
+2, 1738831770112, CLICK
+15, 1738828407736, CLICK
+11, 1738830745280, CLICK
+2, 1738828723460, CLICK
+6, 1738830514687, CLICK
+7, 1738828813139, CLICK
+2, 1738832396151, CLICK
+15, 1738829287999, CLICK
+13, 1738829054160, CLICK
+4, 1738832931698, CLICK
+14, 1738831552880, CLICK
+14, 1738830319038, CLICK
+5, 1738830199986, CLICK
+19, 1738832549452, CLICK
+2, 1738830492671, CLICK
+2, 1738829477828, CLICK
+15, 1738829654244, CLICK
+9, 1738828313606, CLICK
+13, 1738832661490, CLICK
+3, 1738832124274, CLICK
+15, 1738829195276, CLICK
+15, 1738830203765, CLICK
+17, 1738828485440, CLICK
+3, 1738832851600, CLICK
+15, 1738829333276, CLICK
+2, 1738830155117, CLICK
+6, 1738832898435, CLICK
+12, 1738827875882, CLICK
+1, 1738829733255, CLICK
+18, 1738830765472, CLICK
+11, 1738827680769, CLICK
+1, 1738830543946, CLICK
+7, 1738829371278, CLICK
+15, 1738831957745, CLICK
+10, 1738832865298, CLICK
+10, 1738828299443, CLICK
+8, 1738828513664, CLICK
+4, 1738832707258, CLICK
+6, 1738832642882, CLICK
+3, 1738831028580, CLICK
+16, 1738828842826, CLICK
+8, 1738829782192, CLICK
+11, 1738827854142, CLICK
+9, 1738829724691, CLICK
+10, 1738829082507, CLICK
+11, 1738827649853, CLICK
+20, 1738828743139, CLICK
+20, 1738831595987, CLICK
+17, 1738829714592, CLICK
+5, 1738827696432, CLICK
+8, 1738830838530, CLICK
+13, 1738831073694, CLICK
+18, 1738829816372, CLICK
+15, 1738832942753, CLICK
+12, 1738832595845, CLICK
+14, 1738832285710, CLICK
+17, 1738831023337, CLICK
+7, 1738829171949, CLICK
+16, 1738830223478, CLICK
+20, 1738832214595, CLICK
+7, 1738830239961, CLICK
+1, 1738832395165, CLICK
+16, 1738832492859, CLICK
+2, 1738830792548, CLICK
+8, 1738829918626, CLICK
+16, 1738828525578, CLICK
+13, 1738829922601, CLICK
+17, 1738827635465, CLICK
+3, 1738829274923, CLICK
+11, 1738831518566, CLICK
+6, 1738828695770, CLICK
+2, 1738829175165, CLICK
+18, 1738832055536, CLICK
+15, 1738831661467, CLICK
+17, 1738832940597, CLICK
+3, 1738830543758, CLICK
+2, 1738831859528, CLICK
+4, 1738828777977, CLICK
+1, 1738829625831, CLICK
+13, 1738831255149, CLICK
+4, 1738827910329, CLICK
+8, 1738828710729, CLICK
+14, 1738827896665, CLICK
+11, 1738831007207, CLICK
+18, 1738828859935, CLICK
+14, 1738830866569, CLICK
+19, 1738832478907, CLICK
+18, 1738830514437, CLICK
+15, 1738828663724, CLICK
+3, 1738831117037, CLICK
+10, 1738831014103, CLICK
+17, 1738828183080, CLICK
+9, 1738828122279, CLICK
+4, 1738831876049, CLICK
+20, 1738828672172, CLICK
+8, 1738832674119, CLICK
+20, 1738828586549, CLICK
+8, 1738830625788, CLICK
+10, 1738829877886, CLICK
+13, 1738828038951, CLICK
+18, 1738830579984, CLICK
+5, 1738832369757, CLICK
+9, 1738828759332, CLICK
+2, 1738832148464, CLICK
+6, 1738832211683, CLICK
+2, 1738832468389, CLICK
+6, 1738832840033, CLICK
+3, 1738829546938, CLICK
+17, 1738832516049, CLICK
+2, 1738832182192, CLICK
+11, 1738828716413, CLICK
+5, 1738831187417, CLICK
+4, 1738832949243, CLICK
+12, 1738832429980, CLICK
+5, 1738831066442, CLICK
+9, 1738832630173, CLICK
+4, 1738831785994, CLICK
+1, 1738829268559, CLICK
+9, 1738830795687, CLICK
+11, 1738829802265, CLICK
+3, 1738828244568, CLICK
+3, 1738830130979, CLICK
+7, 1738830059326, CLICK
+8, 1738829862702, CLICK
+10, 1738830716229, CLICK
+9, 1738831208000, CLICK
+18, 1738828645343, CLICK
+15, 1738828980138, CLICK
+11, 1738827714413, CLICK
+2, 1738828797339, CLICK
+15, 1738832810974, CLICK
+18, 1738830660538, CLICK
+15, 1738830836722, CLICK
+9, 1738832974735, CLICK
+9, 1738828815210, CLICK
+1, 1738832406258, CLICK
+15, 1738830412169, CLICK
+20, 1738828775538, CLICK
+6, 1738829708560, CLICK
+16, 1738832631861, CLICK
+7, 1738830051482, CLICK
+9, 1738829402975, CLICK
+10, 1738830013532, CLICK
+4, 1738829171221, CLICK
+14, 1738828248262, CLICK
+8, 1738831870240, CLICK
+12, 1738828605916, CLICK
+20, 1738828417121, CLICK
+14, 1738830722350, CLICK
+13, 1738832122631, CLICK
+4, 1738832705633, CLICK
+8, 1738830955260, CLICK
+10, 1738828308489, CLICK
+10, 1738828762716, CLICK
+17, 1738832556388, CLICK
+20, 1738831514083, CLICK
+17, 1738830349804, CLICK
+1, 1738831472284, CLICK
+9, 1738832086991, CLICK
+13, 1738828772758, CLICK
+10, 1738828822325, CLICK
+10, 1738831839561, CLICK
+8, 1738829300078, CLICK
+15, 1738827938946, CLICK
+19, 1738830645586, CLICK
+1, 1738832166474, CLICK
+7, 1738830384509, CLICK
+11, 1738830311464, CLICK
+13, 1738831077054, CLICK
+20, 1738832195024, CLICK
+12, 1738832934769, CLICK
+12, 1738830763289, CLICK
+10, 1738831521286, CLICK
+8, 1738827670150, CLICK
+1, 1738831820637, CLICK
+1, 1738830046334, CLICK
+7, 1738832148626, CLICK
+3, 1738828629484, CLICK
+12, 1738832420876, CLICK
+5, 1738831911911, CLICK
+7, 1738829859189, CLICK
+5, 1738829970514, CLICK
+4, 1738829050144, CLICK
+19, 1738830696173, CLICK
+17, 1738829980725, CLICK
+15, 1738829636501, CLICK
+3, 1738828265926, CLICK
+9, 1738832847608, CLICK
+19, 1738830203523, CLICK
+11, 1738827972748, CLICK
+2, 1738829088493, CLICK
+3, 1738829220694, CLICK
+3, 1738828891408, CLICK
+18, 1738832290744, CLICK
+5, 1738830276244, CLICK
+5, 1738830911996, CLICK
+4, 1738832283834, CLICK
+6, 1738830040097, CLICK
+3, 1738828267149, CLICK
+12, 1738828159563, CLICK
+19, 1738830616124, CLICK
+5, 1738828037792, CLICK
+3, 1738832902036, CLICK
+9, 1738829713465, CLICK
+6, 1738830011629, CLICK
+3, 1738827834919, CLICK
+16, 1738828964139, CLICK
+11, 1738829321852, CLICK
+6, 1738832436719, CLICK
+3, 1738831472243, CLICK
+3, 1738827808725, CLICK
+8, 1738832987492, CLICK
+2, 1738832975283, CLICK
+3, 1738830463336, CLICK
+3, 1738828382758, CLICK
+11, 1738830847963, CLICK
+13, 1738831285396, CLICK
+11, 1738829320452, CLICK
+5, 1738832506583, CLICK
+18, 1738830103497, CLICK
+20, 1738827630962, CLICK
+7, 1738831206189, CLICK
+14, 1738830526798, CLICK
+12, 1738830872048, CLICK
+7, 1738829029363, CLICK
+8, 1738832022871, CLICK
+17, 1738827656871, CLICK
+10, 1738829961024, CLICK
+16, 1738831302756, CLICK
+13, 1738831705460, CLICK
+3, 1738828559624, CLICK
+18, 1738827918637, CLICK
+3, 1738829677443, CLICK
+9, 1738827906269, CLICK
+11, 1738832650447, CLICK
+10, 1738829994643, CLICK
+4, 1738829669987, CLICK
+10, 1738828705859, CLICK
+18, 1738831195510, CLICK
+20, 1738829543694, CLICK
+16, 1738828090773, CLICK
+20, 1738830038444, CLICK
+11, 1738832246987, CLICK
+8, 1738827867250, CLICK
+10, 1738831807071, CLICK
+11, 1738828996208, CLICK
+7, 1738829833431, CLICK
+2, 1738831114327, CLICK
+2, 1738831550259, CLICK
+15, 1738831618138, CLICK
+7, 1738829147521, CLICK
+3, 1738832369100, CLICK
+10, 1738829739154, CLICK
+10, 1738830908137, CLICK
+4, 1738829559659, CLICK
+13, 1738832455444, CLICK
+6, 1738828575656, CLICK
+7, 1738829649474, CLICK
+1, 1738827657431, CLICK
+4, 1738830775727, CLICK
+18, 1738828304179, CLICK
+20, 1738829463052, CLICK
+10, 1738829921226, CLICK
+16, 1738830299221, CLICK
+11, 1738831469933, CLICK
+2, 1738829809842, CLICK
+3, 1738830328430, CLICK
+6, 1738832627995, CLICK
+14, 1738829469959, CLICK
+11, 1738832977043, CLICK
+18, 1738829961822, CLICK
+14, 1738828916106, CLICK
+17, 1738831542158, CLICK
+3, 1738829925834, CLICK
+16, 1738832307916, CLICK
+2, 1738828283582, CLICK
+8, 1738827871399, CLICK
+9, 1738829132504, CLICK
+12, 1738832541592, CLICK
+17, 1738832861616, CLICK
+2, 1738832445281, CLICK
+18, 1738828390967, CLICK
+6, 1738832968068, CLICK
+9, 1738832833404, CLICK
+20, 1738829894544, CLICK
+17, 1738828435933, CLICK
+2, 1738827691482, CLICK
+19, 1738828597600, CLICK
+13, 1738831370267, CLICK
+4, 1738831219855, CLICK
+14, 1738827949004, CLICK
+10, 1738830647370, CLICK
+1, 1738832713767, CLICK
+8, 1738831527705, CLICK
+5, 1738832431142, CLICK
+12, 1738830567932, CLICK
+11, 1738830443609, CLICK
+17, 1738830439936, CLICK
+19, 1738829365141, CLICK
+6, 1738829963700, CLICK
+15, 1738831978552, CLICK
+15, 1738830358399, CLICK
+12, 1738828977485, CLICK
+11, 1738832439447, CLICK
+13, 1738828290601, CLICK
+12, 1738827658955, CLICK
+14, 1738832325445, CLICK
+11, 1738829822922, CLICK
+7, 1738831391268, CLICK
+12, 1738831860979, CLICK
+11, 1738830343749, CLICK
+2, 1738830016675, CLICK
+2, 1738830936365, CLICK
+5, 1738831500152, CLICK
+4, 1738830328019, CLICK
+1, 1738830297700, CLICK
+8, 1738830721674, CLICK
+15, 1738829781459, CLICK
+19, 1738827873738, CLICK
+1, 1738831987069, CLICK
+8, 1738832169276, CLICK
+13, 1738828056878, CLICK
+3, 1738831683148, CLICK
+16, 1738830212689, CLICK
+4, 1738829816146, CLICK
+2, 1738830820843, CLICK
+10, 1738830471542, CLICK
\ No newline at end of file
diff --git a/flink-examples/flink-examples-streaming/src/main/resources/datas/dsv2/join/JoinGrades.csv b/flink-examples/flink-examples-streaming/src/main/resources/datas/dsv2/join/JoinGrades.csv
new file mode 100644
index 00000000000..5988f7509b7
--- /dev/null
+++ b/flink-examples/flink-examples-streaming/src/main/resources/datas/dsv2/join/JoinGrades.csv
@@ -0,0 +1,179 @@
+7,Roland
+9,Brad
+5,Josh
+7,Jim
+6,Jaxon
+5,Aaron
+7,Kevin
+9,Vance
+10,Zayden
+5,Dante
+9,Dave
+6,Kane
+8,Nathan
+11,Sawyer
+7,Peter
+6,Paul
+11,Sofia
+9,Chad
+8,Tanner
+8,Jerry
+8,Miles
+6,Brian
+8,Bob
+7,Kaden
+5,Luna
+6,Kyle
+11,Henry
+5,Tessa
+7,Zane
+6,Randy
+5,Jeremy
+9,Bill
+8,Milo
+11,Jett
+11,Usher
+6,Jack
+7,Daniel
+11,Finn
+5,Tobias
+11,Weston
+8,Hazel
+6,Isla
+8,Rick
+8,Zara
+11,Claire
+7,Walter
+9,Maverick
+6,Yannick
+11,Nico
+7,Charlotte
+11,Willow
+7,Lila
+11,Jake
+11,Leon
+10,Owen
+7,Dustin
+7,Victor
+6,Xander
+8,Zach
+9,Kate
+6,Matt
+11,Evan
+10,Gavin
+6,Grace
+5,Hugh
+5,Mia
+7,Sullivan
+9,Cole
+6,Blake
+9,Grayson
+8,Opal
+7,Piper
+9,Liam
+5,Ella
+11,Avery
+6,Gia
+9,Ira
+7,Quincy
+8,Orion
+9,Ben
+10,Scott
+7,Yusuf
+7,Will
+7,Noah
+10,Scarlett
+11,Quinn
+10,Tyler
+8,Harrison
+6,Francis
+11,Gage
+7,Ivy
+7,Pete
+8,Phoenix
+9,Violet
+7,Riley
+8,Zinnia
+10,Brooklyn
+5,Ximena
+9,Nora
+10,Abel
+9,Ruby
+6,Faith
+11,Wesley
+11,Elias
+7,Zeke
+5,Charles
+8,Andy
+5,Graham
+6,Mason
+11,Olivia
+9,Ray
+5,Ethan
+7,Caleb
+8,John
+11,Adam
+5,Frank
+7,Freya
+6,Paige
+9,Sam
+6,Vaughn
+6,Amelia
+5,Hannah
+11,Victoria
+5,Quentin
+10,Oscar
+6,Ryder
+5,Carter
+10,Odin
+11,Steve
+9,Tom
+9,George
+9,Tim
+9,Bella
+7,Mark
+7,Derek
+6,Julia
+6,Aiden
+6,Wyatt
+11,Emily
+10,Jasmine
+10,Levi
+6,Tristan
+6,Shawn
+11,Dan
+5,Rob
+9,Alex
+7,Silas
+5,Kylie
+7,Nash
+9,Justin
+8,Uriah
+11,Chris
+9,Yara
+9,Alfred
+10,Mike
+10,Andrew
+9,Daisy
+7,Zachary
+5,Yosef
+11,Isaiah
+7,Willa
+11,Trevor
+7,Parker
+11,Eli
+9,Nolan
+7,Luca
+7,Uma
+7,Maya
+11,Yanis
+6,Delilah
+7,Barney
+11,Dylan
+5,Freddie
+8,Ian
+8,Tilly
+10,Xavier
+7,Naomi
+10,Hudson
+6,Rosie
\ No newline at end of file
diff --git a/flink-examples/flink-examples-streaming/src/main/resources/datas/dsv2/join/JoinSalaries.csv b/flink-examples/flink-examples-streaming/src/main/resources/datas/dsv2/join/JoinSalaries.csv
new file mode 100644
index 00000000000..81320ec88be
--- /dev/null
+++ b/flink-examples/flink-examples-streaming/src/main/resources/datas/dsv2/join/JoinSalaries.csv
@@ -0,0 +1,179 @@
+Harrison,45323
+Jerry,60419
+Jeremy,47600
+Gia,53468
+Carter,86160
+Ruby,93835
+John,30101
+Hannah,50344
+Victor,84290
+Sullivan,76711
+Henry,56382
+Barney,86210
+Yanis,53471
+Tessa,40630
+Avery,85037
+Isaiah,28639
+Parker,44682
+Tom,65020
+Scarlett,43623
+Will,46197
+Ivy,36925
+Justin,47400
+Caleb,28402
+Steve,77252
+Victoria,57109
+Delilah,93949
+Brian,83113
+Rosie,25205
+Luna,80556
+Tanner,95946
+Cole,55934
+Randy,95565
+George,54063
+Zara,40436
+Andy,77677
+Elias,29337
+Daisy,76711
+Odin,96975
+Zayden,95989
+Tobias,25205
+Luca,98956
+Willa,82264
+Freya,55685
+Charlotte,78344
+Naomi,95717
+Chad,56521
+Liam,31430
+Adam,30899
+Roland,83653
+Tyler,84894
+Eli,69278
+Phoenix,37820
+Ian,49250
+Paul,81860
+Shawn,59402
+Grace,86319
+Josh,60140
+Zach,74722
+Sawyer,59316
+Walter,36157
+Zane,26808
+Ira,55260
+Tilly,32149
+Andrew,76999
+Uma,68700
+Uriah,86248
+Tim,84904
+Mason,89776
+Yara,77998
+Ethan,76579
+Yannick,45875
+Mike,44094
+Riley,84166
+Trevor,42359
+Nash,38685
+Alfred,93177
+Faith,66807
+Wyatt,31164
+Xavier,33782
+Ella,48817
+Noah,25375
+Dan,61326
+Nora,93132
+Evan,43069
+Frank,72570
+Kevin,90420
+Chris,66051
+Miles,44444
+Xander,76549
+Jasmine,52036
+Ben,72727
+Paige,81647
+Ray,39163
+Bill,51775
+Nico,73944
+Oscar,84544
+Aaron,38182
+Maya,29524
+Zachary,75360
+Yosef,75167
+Rick,48644
+Amelia,25046
+Jack,68636
+Blake,70200
+Quentin,25382
+Quincy,71122
+Gage,93884
+Weston,57736
+Kaden,80364
+Jett,95962
+Owen,60845
+Nathan,85733
+Daniel,96880
+Jim,92116
+Hudson,31625
+Kylie,35255
+Julia,25413
+Dustin,98803
+Ximena,49612
+Grayson,36305
+Mark,83955
+Willow,86070
+Olivia,27185
+Brooklyn,59916
+Aiden,25281
+Orion,79595
+Jake,54117
+Milo,83210
+Matt,91040
+Graham,27215
+Leon,94216
+Zeke,62277
+Zinnia,38802
+Wesley,35434
+Violet,96795
+Peter,58378
+Yusuf,62362
+Kate,41317
+Piper,78504
+Nolan,99367
+Gavin,34774
+Abel,58254
+Charles,83254
+Derek,81831
+Usher,98730
+Dylan,33219
+Francis,70869
+Sofia,93870
+Bella,88917
+Claire,83107
+Vaughn,56677
+Rob,81822
+Hugh,52409
+Vance,81998
+Silas,58175
+Bob,99701
+Opal,36385
+Jaxon,85523
+Isla,92113
+Freddie,99227
+Maverick,74247
+Tristan,64506
+Mia,97159
+Ryder,57147
+Kane,27150
+Dante,58355
+Finn,96828
+Levi,94820
+Scott,25392
+Lila,31072
+Quinn,72312
+Emily,30967
+Brad,27294
+Alex,84775
+Kyle,30488
+Pete,31804
+Dave,79050
+Sam,88896
+Hazel,33264
\ No newline at end of file
diff --git a/flink-examples/flink-examples-streaming/src/main/resources/datas/dsv2/windowing/CountProductSalesWindowingOrders.csv b/flink-examples/flink-examples-streaming/src/main/resources/datas/dsv2/windowing/CountProductSalesWindowingOrders.csv
new file mode 100644
index 00000000000..ce4f7ef36d2
--- /dev/null
+++ b/flink-examples/flink-examples-streaming/src/main/resources/datas/dsv2/windowing/CountProductSalesWindowingOrders.csv
@@ -0,0 +1,500 @@
+5,1738827623135
+13,1738827695047
+17,1738827765340
+19,1738827795117
+20,1738827808798
+10,1738827878429
+10,1738827912673
+12,1738827932765
+14,1738827937414
+11,1738827962181
+8,1738828043872
+8,1738828050320
+5,1738828081514
+15,1738828119020
+20,1738828127328
+6,1738828139412
+6,1738828145051
+20,1738828154771
+15,1738828155109
+20,1738828177448
+11,1738828198462
+2,1738828361260
+4,1738828364684
+12,1738828379629
+7,1738828490241
+6,1738828574672
+12,1738828604164
+11,1738828612738
+15,1738828638954
+15,1738828641169
+6,1738828679474
+1,1738828679519
+16,1738828699835
+17,1738828703919
+11,1738828721416
+10,1738828832278
+9,1738828834560
+12,1738828865599
+3,1738828866004
+5,1738828896110
+3,1738828945948
+1,1738829060511
+8,1738829131922
+2,1738829159935
+2,1738829168222
+11,1738829178220
+3,1738829219983
+8,1738829222869
+8,1738829230143
+18,1738829298372
+19,1738829330325
+6,1738829423248
+12,1738829456528
+19,1738829474918
+5,1738829494451
+19,1738829532216
+14,1738829534838
+11,1738829671662
+7,1738829762829
+19,1738829786681
+14,1738829792357
+16,1738829807634
+9,1738830060700
+16,1738830136188
+10,1738830143106
+16,1738830202172
+11,1738830221830
+13,1738830230002
+8,1738830247299
+4,1738830254684
+8,1738830283510
+20,1738830300960
+11,1738830359332
+16,1738830368809
+14,1738830393276
+11,1738830402452
+4,1738830403144
+5,1738830438069
+18,1738830520601
+15,1738830524718
+3,1738830545820
+4,1738830625772
+4,1738830693332
+9,1738830700879
+6,1738830743748
+17,1738830757655
+13,1738830761668
+7,1738830786986
+9,1738830878036
+4,1738830888052
+10,1738830927629
+17,1738830948796
+20,1738830977362
+8,1738831048677
+3,1738831071799
+15,1738831141455
+6,1738831184716
+8,1738831219872
+7,1738831240501
+4,1738831283437
+3,1738831321520
+1,1738831392927
+13,1738831419144
+15,1738831422019
+17,1738831425466
+7,1738831439034
+10,1738831490898
+20,1738831556265
+1,1738831573472
+15,1738831579294
+20,1738831681486
+7,1738831687349
+15,1738831693973
+5,1738831714942
+7,1738831903381
+14,1738831903728
+14,1738831948546
+13,1738832010604
+5,1738832052084
+3,1738832068061
+13,1738832099442
+2,1738832102939
+9,1738832173522
+15,1738832229074
+11,1738832292360
+11,1738832294220
+19,1738832348660
+9,1738832357612
+20,1738832373835
+12,1738832505432
+11,1738832546783
+2,1738832548152
+16,1738832569172
+10,1738832610737
+20,1738832623255
+14,1738832629391
+19,1738832731271
+13,1738832745454
+2,1738832829990
+17,1738832832253
+14,1738832843381
+3,1738832844733
+11,1738832875933
+15,1738832877740
+7,1738832945598
+15,1738832946703
+18,1738832987659
+17,1738832995211
+4,1738832998989
+7,1738833020803
+11,1738833043131
+12,1738833132994
+4,1738833204170
+19,1738833205064
+16,1738833233436
+14,1738833287088
+19,1738833296437
+14,1738833343063
+16,1738833405305
+20,1738833420091
+1,1738833446911
+4,1738833450068
+11,1738833480296
+10,1738833518087
+18,1738833535521
+7,1738833547365
+4,1738833586069
+15,1738833642250
+19,1738833681750
+6,1738833708465
+7,1738833712801
+15,1738833777120
+20,1738833789401
+8,1738833804888
+4,1738833820849
+7,1738833867397
+17,1738833880026
+2,1738833915868
+9,1738833970333
+2,1738833972273
+14,1738833985139
+1,1738834012488
+7,1738834032193
+7,1738834047315
+1,1738834048396
+8,1738834198820
+7,1738834217923
+3,1738834260863
+5,1738834264165
+14,1738834301884
+7,1738834452264
+18,1738834473174
+7,1738834506017
+20,1738834507527
+10,1738834575303
+13,1738834597385
+2,1738834614934
+8,1738834704730
+20,1738834743589
+7,1738834807881
+5,1738834901651
+1,1738834961775
+2,1738834964282
+19,1738835006924
+11,1738835020500
+19,1738835083420
+3,1738835121098
+11,1738835139306
+7,1738835257877
+11,1738835293622
+13,1738835303685
+10,1738835310085
+9,1738835312292
+6,1738835331004
+9,1738835376290
+15,1738835392381
+11,1738835441780
+10,1738835486033
+4,1738835507487
+7,1738835512449
+13,1738835531613
+12,1738835642944
+2,1738835726395
+19,1738835749089
+16,1738835768656
+9,1738835770214
+15,1738835775178
+6,1738835793838
+11,1738835810503
+6,1738835846938
+12,1738835897220
+4,1738835915270
+8,1738835997588
+10,1738835998077
+3,1738836086804
+11,1738836168870
+13,1738836208185
+12,1738836238081
+2,1738836262794
+19,1738836279280
+15,1738836281844
+8,1738836313140
+16,1738836336273
+10,1738836348340
+6,1738836349360
+20,1738836366483
+12,1738836380134
+11,1738836381710
+18,1738836384984
+8,1738836413884
+8,1738836465166
+2,1738836557114
+4,1738836570895
+19,1738836585743
+19,1738836603107
+2,1738836696057
+2,1738836699857
+15,1738836719895
+19,1738836739807
+19,1738836742573
+17,1738836813704
+14,1738836830018
+6,1738836848841
+20,1738836873520
+12,1738836886035
+2,1738836913580
+6,1738836960155
+17,1738836967205
+17,1738837033837
+2,1738837041748
+14,1738837051325
+16,1738837051446
+17,1738837056625
+1,1738837062117
+11,1738837068519
+7,1738837101624
+5,1738837134713
+19,1738837191678
+17,1738837206021
+5,1738837228241
+2,1738837249502
+7,1738837278187
+7,1738837306536
+3,1738837313032
+1,1738837316373
+14,1738837345387
+7,1738837369304
+7,1738837436834
+10,1738837442455
+7,1738837463956
+15,1738837599962
+10,1738837605481
+4,1738837674355
+20,1738837703869
+14,1738837711380
+7,1738837752250
+13,1738837802132
+8,1738837848716
+16,1738837988770
+2,1738838013637
+3,1738838035820
+15,1738838036999
+2,1738838059112
+16,1738838089886
+18,1738838114530
+12,1738838222874
+2,1738838244748
+8,1738838247603
+6,1738838269611
+3,1738838276977
+16,1738838314638
+5,1738838342436
+1,1738838351035
+18,1738838371125
+19,1738838532438
+17,1738838629329
+11,1738838714191
+7,1738838754330
+8,1738838803154
+17,1738838820150
+5,1738838844228
+20,1738838866272
+18,1738838906632
+12,1738838908976
+4,1738838927666
+3,1738838935746
+2,1738838949472
+17,1738839080905
+14,1738839126323
+17,1738839153702
+5,1738839180523
+1,1738839256576
+10,1738839291243
+7,1738839323787
+1,1738839324333
+3,1738839331059
+18,1738839331375
+6,1738839355134
+5,1738839374685
+6,1738839418430
+10,1738839438559
+10,1738839490029
+12,1738839517410
+12,1738839562876
+4,1738839581448
+9,1738839644574
+4,1738839709843
+2,1738839827794
+7,1738839873797
+8,1738839891305
+20,1738839925725
+9,1738839950898
+3,1738839962676
+1,1738840156795
+15,1738840200158
+1,1738840217577
+20,1738840222762
+2,1738840257449
+7,1738840272091
+16,1738840296772
+1,1738840313126
+16,1738840353822
+9,1738840398637
+9,1738840486972
+7,1738840505587
+14,1738840515530
+3,1738840515534
+13,1738840539668
+9,1738840592359
+12,1738840674437
+6,1738840697081
+11,1738840711854
+15,1738840827429
+4,1738840839364
+19,1738840873998
+10,1738840957485
+20,1738841015870
+10,1738841058617
+18,1738841058960
+10,1738841225923
+7,1738841249336
+8,1738841321459
+12,1738841347775
+17,1738841364005
+7,1738841369342
+3,1738841404176
+1,1738841413265
+8,1738841481845
+2,1738841604198
+10,1738841624591
+9,1738841629913
+6,1738841645636
+3,1738841665539
+15,1738841733871
+5,1738841839540
+3,1738841847807
+2,1738841868619
+12,1738841883957
+13,1738841886953
+1,1738841902504
+20,1738841951440
+8,1738841951584
+14,1738841956601
+8,1738842010181
+13,1738842033704
+18,1738842036769
+4,1738842089555
+1,1738842094684
+16,1738842109984
+19,1738842110819
+6,1738842150141
+1,1738842152533
+17,1738842200817
+2,1738842209200
+16,1738842222255
+4,1738842250540
+16,1738842252403
+10,1738842291304
+17,1738842299492
+1,1738842306040
+20,1738842354983
+12,1738842368788
+19,1738842410844
+1,1738842421913
+2,1738842423593
+12,1738842429170
+9,1738842499690
+5,1738842657182
+9,1738842669593
+17,1738842673086
+18,1738842677450
+15,1738842737337
+2,1738842751255
+14,1738842812183
+20,1738842873203
+13,1738842878865
+1,1738842900678
+10,1738842927732
+1,1738842980151
+16,1738843081742
+14,1738843170722
+20,1738843209606
+18,1738843262321
+1,1738843267788
+11,1738843501686
+19,1738843652439
+10,1738843704759
+13,1738843772878
+11,1738843867874
+12,1738843876609
+19,1738843916049
+7,1738843916124
+18,1738843955295
+9,1738843967525
+17,1738844036458
+2,1738844044937
+11,1738844102092
+4,1738844168619
+11,1738844275621
+6,1738844276939
+14,1738844291106
+20,1738844340255
+5,1738844360787
+4,1738844368818
+13,1738844452194
+5,1738844509473
+15,1738844535067
+7,1738844552431
+16,1738844565915
+11,1738844641781
+4,1738844653403
+8,1738844663768
+20,1738844730628
+10,1738844783876
+11,1738844798583
+7,1738844799792
+16,1738844852656
+9,1738844861612
+8,1738844880825
+5,1738844892913
+20,1738844919769
+10,1738844936097
+16,1738844981027
+4,1738844981634
+18,1738845013957
+5,1738845045507
+6,1738845101655
+5,1738845105558
+17,1738845115206
+7,1738845169940
+10,1738845180062
+9,1738845232814
+16,1738845260341
+1,1738845274881
+4,1738845286656
+19,1738845366037
+1,1738845498453
+11,1738845537130
+15,1738845596379
+13,1738845602543
\ No newline at end of file
diff --git a/flink-examples/flink-examples-streaming/src/test/java/org/apache/flink/streaming/test/examples/DSv2ExamplesITCase.java b/flink-examples/flink-examples-streaming/src/test/java/org/apache/flink/streaming/test/examples/DSv2ExamplesITCase.java
new file mode 100644
index 00000000000..2254ae5c28e
--- /dev/null
+++ b/flink-examples/flink-examples-streaming/src/test/java/org/apache/flink/streaming/test/examples/DSv2ExamplesITCase.java
@@ -0,0 +1,104 @@
+/*
+ * 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.test.examples;
+
+import org.apache.flink.streaming.examples.dsv2.eventtime.CountNewsClicks;
+import org.apache.flink.test.testdata.WordCountData;
+import org.apache.flink.test.util.AbstractTestBase;
+
+import org.junit.jupiter.api.Test;
+
+import java.io.BufferedReader;
+import java.io.FileReader;
+import java.io.IOException;
+import java.net.URL;
+import java.util.Objects;
+
+import static org.apache.flink.test.util.TestBaseUtils.compareResultsByLinesInMemory;
+
+/** Integration test for DataStream API V2 examples. */
+class DSv2ExamplesITCase extends AbstractTestBase {
+
+ @Test
+ void testWordCount() throws Exception {
+ final String textPath = createTempFile("text.txt", WordCountData.TEXT);
+ final String resultPath = getTempDirPath("result");
+
+ org.apache.flink.streaming.examples.dsv2.wordcount.WordCount.main(
+ new String[] {
+ "--input", textPath,
+ "--output", resultPath
+ });
+
+ compareResultsByLinesInMemory(WordCountData.STREAMING_COUNTS_AS_TUPLES, resultPath);
+ }
+
+ @Test
+ void testJoin() throws Exception {
+ final String resultPath = getTempDirPath("result");
+
+ org.apache.flink.streaming.examples.dsv2.join.Join.main(
+ new String[] {"--output", resultPath});
+
+ compareResultsByLinesInMemory(
+ loadFileContent("datas/dsv2/join/JoinResult.csv"), resultPath);
+ }
+
+ @Test
+ void testCountProductSalesWindowing() throws Exception {
+ final String resultPath = getTempDirPath("result");
+
+ org.apache.flink.streaming.examples.dsv2.windowing.CountProductSalesWindowing.main(
+ new String[] {"--output", resultPath});
+
+ compareResultsByLinesInMemory(
+ loadFileContent("datas/dsv2/windowing/CountProductSalesWindowingResult.csv"),
+ resultPath);
+ }
+
+ @Test
+ void testCountNewsClicks() throws Exception {
+ final String resultPath = getTempDirPath("result");
+
+ CountNewsClicks.main(new String[] {"--output", resultPath});
+
+ compareResultsByLinesInMemory(
+ loadFileContent("datas/dsv2/eventtime/CountNewsClicksResult.csv"), resultPath);
+ }
+
+ /**
+ * Load the content from the specified file path for testing purposes.
+ *
+ * @return the file content
+ */
+ private static String loadFileContent(String relativeFilePathInResources) throws IOException {
+ URL url =
+ DSv2ExamplesITCase.class.getClassLoader().getResource(relativeFilePathInResources);
+ Objects.requireNonNull(url);
+
+ StringBuilder contentBuilder = new StringBuilder();
+ BufferedReader br = new BufferedReader(new FileReader(url.getPath()));
+ String line;
+ while ((line = br.readLine()) != null) {
+ contentBuilder.append(line).append("\n");
+ }
+
+ return contentBuilder.toString();
+ }
+}
diff --git a/flink-examples/flink-examples-streaming/src/test/resources/datas/dsv2/eventtime/CountNewsClicksResult.csv b/flink-examples/flink-examples-streaming/src/test/resources/datas/dsv2/eventtime/CountNewsClicksResult.csv
new file mode 100644
index 00000000000..35798a00a40
--- /dev/null
+++ b/flink-examples/flink-examples-streaming/src/test/resources/datas/dsv2/eventtime/CountNewsClicksResult.csv
@@ -0,0 +1,20 @@
+9,29
+16,25
+5,19
+3,34
+10,28
+14,17
+1,20
+7,25
+13,23
+11,36
+17,25
+6,18
+20,23
+15,20
+12,18
+18,21
+8,20
+4,18
+19,10
+2,25
diff --git a/flink-examples/flink-examples-streaming/src/test/resources/datas/dsv2/join/JoinResult.csv b/flink-examples/flink-examples-streaming/src/test/resources/datas/dsv2/join/JoinResult.csv
new file mode 100644
index 00000000000..fe83234a83c
--- /dev/null
+++ b/flink-examples/flink-examples-streaming/src/test/resources/datas/dsv2/join/JoinResult.csv
@@ -0,0 +1,179 @@
+Roland,7,83653
+Brad,9,27294
+Josh,5,60140
+Jim,7,92116
+Jaxon,6,85523
+Aaron,5,38182
+Kevin,7,90420
+Vance,9,81998
+Zayden,10,95989
+Dante,5,58355
+Dave,9,79050
+Kane,6,27150
+Nathan,8,85733
+Sawyer,11,59316
+Peter,7,58378
+Paul,6,81860
+Sofia,11,93870
+Chad,9,56521
+Tanner,8,95946
+Jerry,8,60419
+Miles,8,44444
+Brian,6,83113
+Bob,8,99701
+Kaden,7,80364
+Luna,5,80556
+Kyle,6,30488
+Henry,11,56382
+Tessa,5,40630
+Zane,7,26808
+Randy,6,95565
+Jeremy,5,47600
+Bill,9,51775
+Milo,8,83210
+Jett,11,95962
+Usher,11,98730
+Jack,6,68636
+Daniel,7,96880
+Finn,11,96828
+Tobias,5,25205
+Weston,11,57736
+Hazel,8,33264
+Isla,6,92113
+Rick,8,48644
+Zara,8,40436
+Claire,11,83107
+Walter,7,36157
+Maverick,9,74247
+Yannick,6,45875
+Nico,11,73944
+Charlotte,7,78344
+Willow,11,86070
+Lila,7,31072
+Jake,11,54117
+Leon,11,94216
+Owen,10,60845
+Dustin,7,98803
+Victor,7,84290
+Xander,6,76549
+Zach,8,74722
+Kate,9,41317
+Matt,6,91040
+Evan,11,43069
+Gavin,10,34774
+Grace,6,86319
+Hugh,5,52409
+Mia,5,97159
+Sullivan,7,76711
+Cole,9,55934
+Blake,6,70200
+Grayson,9,36305
+Opal,8,36385
+Piper,7,78504
+Liam,9,31430
+Ella,5,48817
+Avery,11,85037
+Gia,6,53468
+Ira,9,55260
+Quincy,7,71122
+Orion,8,79595
+Ben,9,72727
+Scott,10,25392
+Yusuf,7,62362
+Will,7,46197
+Noah,7,25375
+Scarlett,10,43623
+Quinn,11,72312
+Tyler,10,84894
+Harrison,8,45323
+Francis,6,70869
+Gage,11,93884
+Ivy,7,36925
+Pete,7,31804
+Phoenix,8,37820
+Violet,9,96795
+Riley,7,84166
+Zinnia,8,38802
+Brooklyn,10,59916
+Ximena,5,49612
+Nora,9,93132
+Abel,10,58254
+Ruby,9,93835
+Faith,6,66807
+Wesley,11,35434
+Elias,11,29337
+Zeke,7,62277
+Charles,5,83254
+Andy,8,77677
+Graham,5,27215
+Mason,6,89776
+Olivia,11,27185
+Ray,9,39163
+Ethan,5,76579
+Caleb,7,28402
+John,8,30101
+Adam,11,30899
+Frank,5,72570
+Freya,7,55685
+Paige,6,81647
+Sam,9,88896
+Vaughn,6,56677
+Amelia,6,25046
+Hannah,5,50344
+Victoria,11,57109
+Quentin,5,25382
+Oscar,10,84544
+Ryder,6,57147
+Carter,5,86160
+Odin,10,96975
+Steve,11,77252
+Tom,9,65020
+George,9,54063
+Tim,9,84904
+Bella,9,88917
+Mark,7,83955
+Derek,7,81831
+Julia,6,25413
+Aiden,6,25281
+Wyatt,6,31164
+Emily,11,30967
+Jasmine,10,52036
+Levi,10,94820
+Tristan,6,64506
+Shawn,6,59402
+Dan,11,61326
+Rob,5,81822
+Alex,9,84775
+Silas,7,58175
+Kylie,5,35255
+Nash,7,38685
+Justin,9,47400
+Uriah,8,86248
+Chris,11,66051
+Yara,9,77998
+Alfred,9,93177
+Mike,10,44094
+Andrew,10,76999
+Daisy,9,76711
+Zachary,7,75360
+Yosef,5,75167
+Isaiah,11,28639
+Willa,7,82264
+Trevor,11,42359
+Parker,7,44682
+Eli,11,69278
+Nolan,9,99367
+Luca,7,98956
+Uma,7,68700
+Maya,7,29524
+Yanis,11,53471
+Delilah,6,93949
+Barney,7,86210
+Dylan,11,33219
+Freddie,5,99227
+Ian,8,49250
+Tilly,8,32149
+Xavier,10,33782
+Naomi,7,95717
+Hudson,10,31625
+Rosie,6,25205
diff --git a/flink-examples/flink-examples-streaming/src/test/resources/datas/dsv2/windowing/CountProductSalesWindowingResult.csv b/flink-examples/flink-examples-streaming/src/test/resources/datas/dsv2/windowing/CountProductSalesWindowingResult.csv
new file mode 100644
index 00000000000..2dff4e98259
--- /dev/null
+++ b/flink-examples/flink-examples-streaming/src/test/resources/datas/dsv2/windowing/CountProductSalesWindowingResult.csv
@@ -0,0 +1,115 @@
+5,1738825200000,2
+5,1738828800000,5
+5,1738832400000,2
+5,1738836000000,6
+5,1738839600000,2
+5,1738843200000,5
+13,1738825200000,1
+13,1738828800000,5
+13,1738832400000,4
+13,1738836000000,2
+13,1738839600000,4
+13,1738843200000,3
+17,1738825200000,2
+17,1738828800000,3
+17,1738832400000,3
+17,1738836000000,9
+17,1738839600000,4
+17,1738843200000,2
+19,1738825200000,1
+19,1738828800000,5
+19,1738832400000,7
+19,1738836000000,7
+19,1738839600000,3
+19,1738843200000,3
+20,1738825200000,4
+20,1738828800000,5
+20,1738832400000,5
+20,1738836000000,4
+20,1738839600000,6
+20,1738843200000,4
+10,1738825200000,2
+10,1738828800000,4
+10,1738832400000,6
+10,1738836000000,6
+10,1738839600000,6
+10,1738843200000,4
+12,1738825200000,3
+12,1738828800000,2
+12,1738832400000,4
+12,1738836000000,7
+12,1738839600000,5
+12,1738843200000,1
+14,1738825200000,1
+14,1738828800000,5
+14,1738832400000,6
+14,1738836000000,5
+14,1738839600000,4
+14,1738843200000,1
+11,1738825200000,4
+11,1738828800000,7
+11,1738832400000,9
+11,1738836000000,4
+11,1738839600000,1
+11,1738843200000,7
+8,1738825200000,2
+8,1738828800000,7
+8,1738832400000,4
+8,1738836000000,6
+8,1738839600000,5
+8,1738843200000,2
+15,1738825200000,4
+15,1738828800000,6
+15,1738832400000,6
+15,1738836000000,4
+15,1738839600000,4
+15,1738843200000,2
+6,1738825200000,4
+6,1738828800000,3
+6,1738832400000,4
+6,1738836000000,6
+6,1738839600000,3
+6,1738843200000,2
+2,1738825200000,1
+2,1738828800000,3
+2,1738832400000,7
+2,1738836000000,11
+2,1738839600000,7
+2,1738843200000,1
+4,1738825200000,1
+4,1738828800000,6
+4,1738832400000,7
+4,1738836000000,4
+4,1738839600000,4
+4,1738843200000,5
+7,1738825200000,1
+7,1738828800000,6
+7,1738832400000,13
+7,1738836000000,9
+7,1738839600000,5
+7,1738843200000,4
+1,1738825200000,1
+1,1738828800000,3
+1,1738832400000,4
+1,1738836000000,5
+1,1738839600000,11
+1,1738843200000,3
+16,1738825200000,1
+16,1738828800000,4
+16,1738832400000,4
+16,1738836000000,5
+16,1738839600000,6
+16,1738843200000,4
+9,1738828800000,6
+9,1738832400000,4
+9,1738839600000,8
+9,1738843200000,3
+3,1738828800000,7
+3,1738832400000,3
+3,1738836000000,6
+3,1738839600000,5
+18,1738828800000,2
+18,1738832400000,3
+18,1738836000000,5
+18,1738839600000,3
+18,1738843200000,3
diff --git a/pom.xml b/pom.xml
index 0903bea943a..e9a388bc7fd 100644
--- a/pom.xml
+++ b/pom.xml
@@ -1635,6 +1635,8 @@ under the License.
<exclude>flink-tests/src/test/resources/avro/*.avsc</exclude>
<exclude>flink-connectors/flink-connector-datagen-test/src/test/resources/avro/*.avsc</exclude>
<exclude>flink-runtime-web/web-dashboard/dev/notice-template</exclude>
+ <exclude>flink-examples/flink-examples-streaming/src/main/resources/datas/**</exclude>
+ <exclude>flink-examples/flink-examples-streaming/src/test/resources/datas/**</exclude>
<!-- ArchUnit violation stores -->
<exclude>**/archunit-violations/**</exclude>