Skip to content

[BP-2.4][FLINK-41407][docs] Fix code examples in the documentation #29896

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
MartijnVisser merged 1 commit into from
Nov 12, 2026
Merged

[BP-2.4][FLINK-41407][docs] Fix code examples in the documentation #29896

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
289 changes: 25 additions & 264 deletions docs/content.zh/docs/concepts/sql-table-concepts/time_attributes.md
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -41,56 +41,18 @@ Flink 可以基于几种不同的 *时间* 概念来处理数据。

像窗口（在 [Table API]({{< ref "docs/dev/table/tableApi" >}}#group-windows) 和 [SQL]({{< ref "docs/sql/reference/queries/window-agg" >}}) ）这种基于时间的操作，需要有时间信息。因此，Table API 中的表就需要提供*逻辑时间属性*来表示时间，以及支持时间相关的操作。

每种类型的表都可以有时间属性，可以在用CREATE TABLE DDL创建表的时候指定、也可以在 `DataStream` 中指定、也可以在定义 `TableSource` 时指定。一旦时间属性定义好，它就可以像普通列一样使用，也可以在时间相关的操作中使用。
每种类型的表都可以有时间属性，可以在用CREATE TABLE DDL创建表的时候指定，也可以在 `DataStream` 中指定。一旦时间属性定义好，它就可以像普通列一样使用，也可以在时间相关的操作中使用。

只要时间属性没有被修改，而是简单地从一个表传递到另一个表，它就仍然是一个有效的时间属性。时间属性可以像普通的时间戳的列一样被使用和计算。一旦时间属性被用在了计算中，它就会被物化，进而变成一个普通的时间戳。普通的时间戳是无法跟 Flink 的时间以及watermark等一起使用的，所以普通的时间戳就无法用在时间相关的操作中。

Table API 程序需要在 streaming environment 中指定时间属性：

{{< tabs "bc88319c-8c8f-4f9d-8972-a2d6efe478d7" >}}
{{< tab "Java" >}}
```java
final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

env.setStreamTimeCharacteristic(TimeCharacteristic.ProcessingTime); // default

// 或者:
// env.setStreamTimeCharacteristic(TimeCharacteristic.IngestionTime);
// env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);
```
{{< /tab >}}
{{< tab "Scala" >}}
```scala
val env = StreamExecutionEnvironment.getExecutionEnvironment

env.setStreamTimeCharacteristic(TimeCharacteristic.ProcessingTime) // default

// 或者:
// env.setStreamTimeCharacteristic(TimeCharacteristic.IngestionTime)
// env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime)
```
{{< /tab >}}
{{< tab "Python" >}}
```python
env = StreamExecutionEnvironment.get_execution_environment()

env.set_stream_time_characteristic(TimeCharacteristic.ProcessingTime) # default

# 或者:
# env.set_stream_time_characteristic(TimeCharacteristic.IngestionTime)
# env.set_stream_time_characteristic(TimeCharacteristic.EventTime)
```
{{< /tab >}}
{{< /tabs >}}

<a name="processing-time"></a>

处理时间
---------------

处理时间是基于机器的本地时间来处理数据，它是最简单的一种时间概念，但是它不能提供确定性。它既不需要从数据里获取时间，也不需要生成 watermark。

共有三种方法可以定义处理时间。
共有两种方法可以定义处理时间。

### 在创建表的 DDL 中定义

Expand All @@ -117,105 +79,26 @@ GROUP BY TUMBLE(user_action_time, INTERVAL '10' MINUTE);

处理时间属性可以在 schema 定义的时候用 `.proctime` 后缀来定义。时间属性一定不能定义在一个已有字段上，所以它只能定义在 schema 定义的最后。

在 Java 中，在传给 `fromDataStream` 的 `Schema` 中将处理时间属性声明为计算列 `PROCTIME()`。

{{< tabs "ec1506ee-adf2-4668-a6f0-7944a5b24bc2" >}}
{{< tab "Java" >}}
```java
DataStream<Tuple2<String, String>> stream = ...;

// 声明一个额外的字段作为时间属性字段
Table table = tEnv.fromDataStream(stream, $("user_name"), $("data"), $("user_action_time").proctime());
// declare an additional column as a processing time attribute
Table table = tEnv.fromDataStream(
stream,
Schema.newBuilder()
.columnByExpression("user_action_time", "PROCTIME()")
.build());

WindowedTable windowedTable = table.window(
GroupWindowedTable windowedTable = table.window(
Tumble.over(lit(10).minutes())
.on($("user_action_time"))
.as("userActionWindow"));
```
{{< /tab >}}
{{< tab "Scala" >}}
```scala
val stream: DataStream[(String, String)] = ...

// 声明一个额外的字段作为时间属性字段
val table = tEnv.fromDataStream(stream, $"UserActionTimestamp", $"user_name", $"data", $"user_action_time".proctime)

val windowedTable = table.window(Tumble over 10.minutes on $"user_action_time" as "userActionWindow")
```
{{< /tab >}}
{{< /tabs >}}

### 使用 TableSource 定义

处理时间属性可以在实现了 `DefinedProctimeAttribute` 的 `TableSource` 中定义。逻辑的时间属性会放在 `TableSource` 已有物理字段的最后

{{< tabs "4b87d0d7-487e-4b11-a01a-4812f6b71e2d" >}}
{{< tab "Java" >}}
```java
// 定义一个由处理时间属性的 table source
public class UserActionSource implements StreamTableSource<Row>, DefinedProctimeAttribute {

@Override
public TypeInformation<Row> getReturnType() {
String[] names = new String[] {"user_name" , "data"};
TypeInformation[] types = new TypeInformation[] {Types.STRING(), Types.STRING()};
return Types.ROW(names, types);
}

@Override
public DataStream<Row> getDataStream(StreamExecutionEnvironment execEnv) {
// create stream
DataStream<Row> stream = ...;
return stream;
}

@Override
public String getProctimeAttribute() {
// 这个名字的列会被追加到最后，作为第三列
return "user_action_time";
}
}

// register table source
tEnv.registerTableSource("user_actions", new UserActionSource());

WindowedTable windowedTable = tEnv
.from("user_actions")
.window(Tumble
.over(lit(10).minutes())
.on($("user_action_time"))
.as("userActionWindow"));
```
{{< /tab >}}
{{< tab "Scala" >}}
```scala
// 定义一个由处理时间属性的 table source
class UserActionSource extends StreamTableSource[Row] with DefinedProctimeAttribute {

override def getReturnType = {
val names = Array[String]("user_name" , "data")
val types = Array[TypeInformation[_]](Types.STRING, Types.STRING)
Types.ROW(names, types)
}

override def getDataStream(execEnv: StreamExecutionEnvironment): DataStream[Row] = {
// create stream
val stream = ...
stream
}

override def getProctimeAttribute = {
// 这个名字的列会被追加到最后，作为第三列
"user_action_time"
}
}

// register table source
tEnv.registerTableSource("user_actions", new UserActionSource)

val windowedTable = tEnv
.from("user_actions")
.window(Tumble over 10.minutes on $"user_action_time" as "userActionWindow")
```
{{< /tab >}}
{{< /tabs >}}

事件时间
Expand All @@ -227,7 +110,7 @@ val windowedTable = tEnv

为了能够处理乱序的事件，并且区分正常到达和晚到的事件，Flink 需要从事件中获取事件时间并且产生 watermark（[watermarks]({{< ref "docs/concepts/time" >}})）。

事件时间属性也有类似于处理时间的三种定义方式：在DDL中定义、在 DataStream 到 Table 转换时定义、用 TableSource 定义。
事件时间属性也有类似于处理时间的两种定义方式：在DDL中定义、在 DataStream 到 Table 转换时定义。

### 在 DDL 中定义

Expand Down Expand Up @@ -370,7 +253,7 @@ select ... from source_table /*+ OPTIONS('scan.watermark.alignment.group'='align
### 在 DataStream 到 Table 转换时定义

事件时间属性可以用 `.rowtime` 后缀在定义 `DataStream` schema 的时候来定义。[时间戳和 watermark]({{< ref "docs/concepts/time" >}}) 在这之前一定是在 `DataStream` 上已经定义好了。
在从 DataStream 转换到 Table 时，由于 `DataStream` 没有时区概念，因此 Flink 总是将 `rowtime` 属性解析成 `TIMESTAMP WITHOUT TIME ZONE` 类型，并且将所有事件时间的值都视为 UTC 时区的值。
使用 `.rowtime` 表达式转换时，由于 `DataStream` 没有时区概念，因此 Flink 将 `rowtime` 属性解析成 `TIMESTAMP WITHOUT TIME ZONE` 类型，并且将所有事件时间的值都视为 UTC 时区的值。

在从 `DataStream` 到 `Table` 转换时定义事件时间属性有两种方式。取决于用 `.rowtime` 后缀修饰的字段名字是否是已有字段，事件时间字段可以是：

Expand All @@ -379,153 +262,31 @@ select ... from source_table /*+ OPTIONS('scan.watermark.alignment.group'='align

不管在哪种情况下，事件时间字段都表示 `DataStream` 中定义的事件的时间戳。

在 Java 中，在传给 `fromDataStream` 的 `Schema` 中声明事件时间属性：`rowtime` 元数据列以 `TIMESTAMP_LTZ(3)` 类型保存每条流记录的时间戳，`SOURCE_WATERMARK()` 复用 `DataStream` 的 watermark。上面两种方式使用基于表达式的转换，这种转换在 Java 中已被弃用。详情请参考 [DataStream API Integration]({{< ref "docs/dev/table/data_stream_api" >}})。

{{< tabs "4ddfd970-e823-48c2-ae38-4e110ada0f9a" >}}
{{< tab "Java" >}}
```java

// Option 1:

// 基于 stream 中的事件产生时间戳和 watermark
// extract timestamp and assign watermarks based on knowledge of the stream
DataStream<Tuple2<String, String>> stream = inputStream.assignTimestampsAndWatermarks(...);

// 声明一个额外的逻辑字段作为事件时间属性
Table table = tEnv.fromDataStream(stream, $("user_name"), $("data"), $("user_action_time").rowtime());


// Option 2:

// 从第一个字段获取事件时间，并且产生 watermark
DataStream<Tuple3<Long, String, String>> stream = inputStream.assignTimestampsAndWatermarks(...);

// 第一个字段已经用作事件时间抽取了，不用再用一个新字段来表示事件时间了
Table table = tEnv.fromDataStream(stream, $("user_action_time").rowtime(), $("user_name"), $("data"));
// declare an additional column that holds the timestamp of each stream record as an
// event time attribute, and use the watermarks of the DataStream
Table table = tEnv.fromDataStream(
stream,
Schema.newBuilder()
.columnByMetadata("user_action_time", DataTypes.TIMESTAMP_LTZ(3), "rowtime")
.watermark("user_action_time", "SOURCE_WATERMARK()")
.build());

// Usage:

WindowedTable windowedTable = table.window(Tumble
GroupWindowedTable windowedTable = table.window(Tumble
.over(lit(10).minutes())
.on($("user_action_time"))
.as("userActionWindow"));
```
{{< /tab >}}
{{< tab "Scala" >}}
```scala

// Option 1:

// 基于 stream 中的事件产生时间戳和 watermark
val stream: DataStream[(String, String)] = inputStream.assignTimestampsAndWatermarks(...)

// 声明一个额外的逻辑字段作为事件时间属性
val table = tEnv.fromDataStream(stream, $"user_name", $"data", $"user_action_time".rowtime)


// Option 2:

// 从第一个字段获取事件时间，并且产生 watermark
val stream: DataStream[(Long, String, String)] = inputStream.assignTimestampsAndWatermarks(...)

// 第一个字段已经用作事件时间抽取了，不用再用一个新字段来表示事件时间了
val table = tEnv.fromDataStream(stream, $"user_action_time".rowtime, $"user_name", $"data")

// Usage:

val windowedTable = table.window(Tumble over 10.minutes on $"user_action_time" as "userActionWindow")
```
{{< /tab >}}
{{< /tabs >}}

### 使用 TableSource 定义

事件时间属性可以在实现了 `DefinedRowTimeAttributes` 的 `TableSource` 中定义。`getRowtimeAttributeDescriptors()` 方法返回 `RowtimeAttributeDescriptor` 的列表，包含了描述事件时间属性的字段名字、如何计算事件时间、以及 watermark 生成策略等信息。

同时需要确保 `getDataStream` 返回的 `DataStream` 已经定义好了时间属性。
只有在定义了 `StreamRecordTimestamp` 时间戳分配器的时候，才认为 `DataStream` 是有时间戳信息的。
只有定义了 `PreserveWatermarks` watermark 生成策略的 `DataStream` 的 watermark 才会被保留。反之，则只有时间字段的值是生效的。

{{< tabs "bfc26d42-c602-402e-98fc-fe6eb787d283" >}}
{{< tab "Java" >}}
```java
// 定义一个有事件时间属性的 table source
public class UserActionSource implements StreamTableSource<Row>, DefinedRowtimeAttributes {

@Override
public TypeInformation<Row> getReturnType() {
String[] names = new String[] {"user_name", "data", "user_action_time"};
TypeInformation[] types =
new TypeInformation[] {Types.STRING(), Types.STRING(), Types.LONG()};
return Types.ROW(names, types);
}

@Override
public DataStream<Row> getDataStream(StreamExecutionEnvironment execEnv) {
// 构造 DataStream
// ...
// 基于 "user_action_time" 定义 watermark
DataStream<Row> stream = inputStream.assignTimestampsAndWatermarks(...);
return stream;
}

@Override
public List<RowtimeAttributeDescriptor> getRowtimeAttributeDescriptors() {
// 标记 "user_action_time" 字段是事件时间字段
// 给 "user_action_time" 构造一个时间属性描述符
RowtimeAttributeDescriptor rowtimeAttrDescr = new RowtimeAttributeDescriptor(
"user_action_time",
new ExistingField("user_action_time"),
new AscendingTimestamps());
List<RowtimeAttributeDescriptor> listRowtimeAttrDescr = Collections.singletonList(rowtimeAttrDescr);
return listRowtimeAttrDescr;
}
}

// register the table source
tEnv.registerTableSource("user_actions", new UserActionSource());

WindowedTable windowedTable = tEnv
.from("user_actions")
.window(Tumble.over(lit(10).minutes()).on($("user_action_time")).as("userActionWindow"));
```
{{< /tab >}}
{{< tab "Scala" >}}
```scala
// 定义一个有事件时间属性的 table source
class UserActionSource extends StreamTableSource[Row] with DefinedRowtimeAttributes {

override def getReturnType = {
val names = Array[String]("user_name" , "data", "user_action_time")
val types = Array[TypeInformation[_]](Types.STRING, Types.STRING, Types.LONG)
Types.ROW(names, types)
}

override def getDataStream(execEnv: StreamExecutionEnvironment): DataStream[Row] = {
// 构造 DataStream
// ...
// 基于 "user_action_time" 定义 watermark
val stream = inputStream.assignTimestampsAndWatermarks(...)
stream
}

override def getRowtimeAttributeDescriptors: util.List[RowtimeAttributeDescriptor] = {
// 标记 "user_action_time" 字段是事件时间字段
// 给 "user_action_time" 构造一个时间属性描述符
val rowtimeAttrDescr = new RowtimeAttributeDescriptor(
"user_action_time",
new ExistingField("user_action_time"),
new AscendingTimestamps)
val listRowtimeAttrDescr = Collections.singletonList(rowtimeAttrDescr)
listRowtimeAttrDescr
}
}

// register the table source
tEnv.registerTableSource("user_actions", new UserActionSource)

val windowedTable = tEnv
.from("user_actions")
.window(Tumble over 10.minutes on $"user_action_time" as "userActionWindow")
```
{{< /tab >}}
{{< /tabs >}}

{{< top >}}
2 changes: 1 addition & 1 deletion docs/content.zh/docs/connectors/datastream/filesystem.md
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -106,7 +106,7 @@ final FileSource<String> source =
{{< tab "Python" >}}
```python
source = FileSource.for_record_stream_format(...) \
.monitor_continously(Duration.of_millis(5)) \
.monitor_continuously(Duration.of_millis(5)) \
.build()
```
{{< /tab >}}
Expand Down
4 changes: 2 additions & 2 deletions docs/content.zh/docs/connectors/datastream/formats/avro.md
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -51,10 +51,10 @@ AvroInputFormat<User> users = new AvroInputFormat<User>(in, User.class);
DataStream<User> usersDS = env.createInput(users);
```

注意，`User` 是一个通过 Avro schema生成的 POJO 类。Flink 还允许选择 POJO 中字符串类型的键。例如：
注意，`User` 是一个通过 Avro schema生成的 POJO 类。你可以通过 key selector 按这些 POJO 的某个字段对流进行 key 分区。例如：

```java
usersDS.keyBy("name");
usersDS.keyBy(user -> user.getName());
```


Expand Down
7 changes: 4 additions & 3 deletions docs/content.zh/docs/connectors/datastream/formats/azure_table_storage.md
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -69,11 +69,12 @@ curl https://flink.apache.org/q/quickstart.sh | bash

```java
import java.util.Map;
import org.apache.flink.api.common.RuntimeExecutionMode;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.java.DataStream;
import org.apache.flink.api.java.ExecutionEnvironment;
import org.apache.flink.api.java.hadoop.mapreduce.HadoopInputFormat;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.hadoopcompatibility.mapreduce.HadoopInputFormat;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import com.microsoft.hadoop.azure.AzureTableConfiguration;
Expand Down
Loading