是的，Flink CDC 支持指定时间消费，可以设置消费起始时间和结束时间，包括对 SQL Server 和 Postgres 的支持。

Flink CDC 支持指定时间消费的功能，可以用于在特定时间范围内消费数据，对于 SQL Server 和 Postgres，可以通过配置来实现指定时间消费。

SQL Server 的指定时间消费

在 Flink CDC 中，可以使用 Debezium 连接器来读取 SQL Server 数据库的变化日志，并通过 Debezium 提供的 include.schema.changes 参数来控制是否包含模式变更事件，可以使用 startup.mode 参数来设置启动模式，以实现指定时间消费。

单元表格：SQL Server 的指定时间消费配置

参数	默认值	说明
include.schema.changes	false	是否包含模式变更事件
startup.mode	latest	启动模式，可以选择 "latest"（最新）或 "specificoffset"（指定偏移量）

示例代码：

import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
import org.apache.flink.table.api.EnvironmentSettings;
import org.apache.flink.table.catalog.debezium.DebeziumOptions;
import org.apache.flink.table.catalog.debezium.DebeziumTableFactory;
import org.apache.flink.table.descriptors.*;
import org.apache.flink.table.sources.RowtimeAttributeDescriptor;
import org.apache.flink.types.RowType;
// ... 创建流处理执行环境、表执行环境等 ...
// 创建源表描述符
String sourceDDL = "CREATE TABLE my_source (...) WITH (...)"; // 根据实际需求填写 DDL
SourceTableDescriptor sourceTableDescriptor = new SourceTableDescriptor(sourceDDL, new RowtimeAttributeDescriptor("ts", "rowtime", "TIMESTAMP(3)"));
// 创建目标表描述符
String sinkDDL = "CREATE TABLE my_sink (...) WITH (...)"; // 根据实际需求填写 DDL
SinkTableDescriptor sinkTableDescriptor = new SinkTableDescriptor(sinkDDL);
// 创建连接器选项并设置启动模式为 latest（最新）或 specificoffset（指定偏移量）
DebeziumOptions options = new DebeziumOptions().withStartupMode(DebeziumOptions.StartupMode.LATEST); // 或者使用其他启动模式
// 注册源表和目标表，并添加连接器选项
StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env, settings);
tableEnv.registerTableSource(sourceTableDescriptor, new DebeziumTableFactory<>(options));
tableEnv.registerTableSink(sinkTableDescriptor);
// ... 执行 Flink SQL 查询或转换操作 ...

Postgres 的指定时间消费

在 Flink CDC 中，同样可以使用 Debezium 连接器来读取 Postgres 数据库的变化日志，并通过 Debezium 提供的 include.schema.changes 参数来控制是否包含模式变更事件，可以使用 startup.mode 参数来设置启动模式，以实现指定时间消费。

单元表格：Postgres 的指定时间消费配置

参数	默认值	说明
include.schema.changes	false	是否包含模式变更事件
startup.mode	latest	启动模式，可以选择 "latest"（最新）或 "specificoffset"（指定偏移量）

示例代码：

import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
import org.apache.flink.table.api.EnvironmentSettings;
import org.apache.flink.table.catalog.debezium.DebeziumOptions;
import org.apache.flink.table.catalog.debezium.DebeziumTableFactory;
import org.apache.flink.table.descriptors.*;
import org.apache.flink.table.sources.*;
import org.apache.flink.types.*;
// ... 创建流处理执行环境、表执行环境等 ...
// 创建源表描述符
String sourceDDL = "CREATE TABLE my_source (...) WITH (...)"; // 根据实际需求填写 DDL
SourceTableDescriptor sourceTableDescriptor = new SourceTableDescriptor(sourceDDL, new RowtimeAttributeDescriptor("ts", "rowtime", "TIMESTAMP(3)"));
// 创建目标表描述符
String sinkDDL = "CREATE TABLE my_sink (...) WITH (...)"; // 根据实际需求填写 DDL
SinkTableDescriptor sinkTableDescriptor = new SinkTableDescriptor(sinkDDL);
// 创建连接器选项并设置启动模式为 latest（最新）或 specificoffset（指定偏移量）
DebeziumOptions options = new DebeziumOptions().withStartupMode(DebeziumOptions.StartupMode

新闻名称：FlinkCDC里sqlserver和postgres会支持指定时间消费吗？
网址分享：http://gydahua.com/article/dhsdpgd.html