create table order_core_order_master ( message string, proctime_kafka_1 as proctime() ) with ( "connector.type" = "kafka", "connector.version" = "0.10", "connector.properties.bootstrap.servers" = "10.188.18.34:9092,10.188.17.128:9092,10.188.18.231:9092,10.188.19.163:9092,10.188.16.210:9092", "connector.properties.group.id" = "flink_159374_order_core_order_master", "connector.properties.flink.partition-discovery.interval-millis" = "30000", "connector.topic" = "order_core_all", "format.type" = "csv", "format.field-delimiter" = "\u0001", "format.line-delimiter" = "\r", "format.quote-character" = "''", -- "format.type" = "json", --"connector.startup-mode" = "latest-offset" "connector.startup-mode" = "timestamp", "connector.startup-timestamp-millis" = "1685653200000" --'2023-06-02 05:00:00' );

这是一个创建名为order_core_order_master的表的SQL语句。该表包含一个名为message的字符串类型的列，以及一个名为proctime_kafka_1的虚拟列，该列使用系统时间作为其值。此外，该表还定义了使用Apache Kafka作为数据源的连接器，指定了Kafka服务器地址和组ID等属性，以及使用CSV格式解析数据。最后，该表还指定了连接器的启动模式为“timestamp”，并指定了一个时间戳作为其启动时间。

相关问题

flink1.11 中 使用SQL 获取 Kafka Connector 时 如何获取或提取到 kafka消息落盘kafka时间

在 Flink 1.11 中，可以使用 Flink 的 Table API 和 SQL API 来从 Kafka 中读取数据，并将其转换为 Table。假设你已经成功地将 Kafka 的数据转换为 Table，那么如何获取消息的落盘时间呢？

Flink 的 Table API 和 SQL API 为每个输入数据源都定义了一个元组类型，其中包含了所有字段的名称和类型。对于 Kafka 的元组类型，通常会包含以下字段：

- `msg`: Kafka 消息的内容。
- `offset`: Kafka 消息的偏移量。
- `partition`: Kafka 消息所在的分区。
- `timestamp`: Kafka 消息的时间戳。

要获取消息的落盘时间，可以使用 Flink SQL 中的 `PROCTIME()` 函数。`PROCTIME()` 函数返回当前处理时间，因此可以将其与 `timestamp` 字段进行比较，以确定消息是否已经落盘。

以下是一个使用 Flink SQL 从 Kafka 中读取数据并获取消息落盘时间的示例：

CREATE TABLE kafkaTable (
  msg STRING,
  offset BIGINT,
  partition INT,
  timestamp TIMESTAMP(3)
) WITH (
  'connector' = 'kafka',
  'topic' = 'myTopic',
  'properties.bootstrap.servers' = 'localhost:9092',
  'scan.startup.mode' = 'earliest-offset',
  'format' = 'json'
);

SELECT
  msg,
  offset,
  partition,
  timestamp,
  PROCTIME() AS processingTime
FROM kafkaTable;

在上面的示例中，我们定义了一个名为 `kafkaTable` 的表，该表从名为 `myTopic` 的 Kafka 主题中读取数据。在 `SELECT` 子句中，我们选择了所有字段，并使用 `PROCTIME()` 函数添加了一个名为 `processingTime` 的新字段。这个新字段包含了当前处理时间，可以与 `timestamp` 字段进行比较。

wts as protime（）什么意思

在 Flink SQL 中，`wts AS PROCTIME()` 表示将一个字段命名为 `wts`，并将其作为处理时间属性（Processing Time）。

处理时间是指 Flink 处理数据的时间，它由 Flink 自动分配给数据记录。使用处理时间进行处理时，不需要在数据中显式地包含时间戳字段，而是由 Flink 在运行时动态地分配处理时间。

通过使用 `AS PROCTIME()`，你可以将一个字段标记为处理时间属性，并在查询中使用它。例如：

SELECT wts, COUNT(*)
FROM MyTable
GROUP BY wts;

在上述示例中，`wts` 被标记为处理时间属性，并在查询中使用它进行分组操作。

需要注意的是，处理时间是相对于 Flink 执行任务的机器时间。在流式处理中，处理时间会随着数据到达 Flink 的顺序而递增。因此，如果你需要基于事件的时间属性进行处理，请使用事件时间（Event Time）而不是处理时间。