Flink CDC 里2.4cdc同步pg比原来时间多8小时，修改什么参数？

在使用 Flink CDC 同步 PostgreSQL 数据时，如果发现同步的时间比原始时间多 8 小时，这通常是由于时区配置不一致导致的。以下是问题的原因分析和解决方案：

问题原因

1. PostgreSQL 的时间类型与时区处理
   PostgreSQL 的 timestamp 类型默认是无时区信息的（TIMESTAMP WITHOUT TIME ZONE），而 timestamptz 类型是带有时区信息的（TIMESTAMP WITH TIME ZONE）。Flink CDC 在读取 PostgreSQL 数据时，会根据连接器的配置和 Flink 的时区设置对时间字段进行解析和转换。

2. Flink 的时区配置
   Flink 默认使用 UTC 时间（UTC+0）来处理时间字段。如果 PostgreSQL 中存储的是无时区的 timestamp 类型数据，Flink CDC 在解析时可能会将其视为 UTC 时间，并根据 Flink 的时区配置（例如 Asia/Shanghai，即 UTC+8）进行转换，从而导致时间偏移 8 小时。

3. PostgreSQL 未显式支持时区参数
   PostgreSQL CDC 连接器本身并未提供类似 MySQL 的 server-time-zone 参数来显式指定时区。因此，时区问题需要通过 Flink 的全局配置或自定义逻辑来解决。

解决方案

方法一：调整 Flink 的时区配置

Flink 提供了 table.local-time-zone 参数，用于指定 Flink 处理时间字段时的时区。如果 PostgreSQL 中的时间字段是无时区的 timestamp 类型，可以通过以下步骤解决时区问题：

1. 修改 Flink 配置文件
   在 Flink 的 flink-conf.yaml 文件中添加以下配置：

   table.local-time-zone: UTC
   

   这将确保 Flink 在处理时间字段时不进行额外的时区转换。

2. 动态设置作业参数
   如果不想修改全局配置，可以在提交 Flink 作业时动态设置时区参数。例如：

   SET 'table.local-time-zone' = 'UTC';
   

3. 验证效果
   确保 Flink 作业在处理 PostgreSQL 数据时，时间字段不会被错误地转换为其他时区。

方法二：在 PostgreSQL 中使用 timestamptz 类型

如果可以修改 PostgreSQL 表结构，建议将时间字段改为带时区的 timestamptz 类型。这样可以避免因时区配置不一致导致的问题：

1. 修改表结构
   将 PostgreSQL 表中的时间字段从 timestamp 修改为 timestamptz：

   ALTER TABLE your_table ALTER COLUMN your_timestamp_column TYPE timestamptz;
   

2. 更新数据
   如果已有数据存储为无时区的 timestamp 类型，需要手动调整数据以匹配正确的时区。例如：

   UPDATE your_table SET your_timestamp_column = your_timestamp_column AT TIME ZONE 'UTC';
   

3. 重新同步
   修改完成后，重新启动 Flink CDC 作业，确保时间字段能够正确解析。

方法三：自定义反序列化逻辑

如果无法修改 PostgreSQL 表结构或 Flink 配置，可以通过自定义反序列化逻辑来处理时间字段的时区问题：

1. 实现自定义反序列化器
   在 Flink CDC 连接器中实现 DebeziumDeserializationSchema 接口，并在解析时间字段时显式指定时区。例如：

   public class CustomDeserializer implements DebeziumDeserializationSchema<RowData> {
      private final ZoneId serverTimeZone = ZoneId.of("UTC");
   
      @Override
      public void deserialize(SourceRecord record, Collector<RowData> out) throws Exception {
          Struct value = (Struct) record.value();
          Timestamp timestamp = value.getTimestamp("your_timestamp_field");
          LocalDateTime localDateTime = timestamp.toLocalDateTime();
          ZonedDateTime zonedDateTime = localDateTime.atZone(serverTimeZone);
          // 转换为 UTC 时间
          long epochMillis = zonedDateTime.toInstant().toEpochMilli();
          // 构造 RowData 并输出
          RowData rowData = ...; // 根据业务逻辑构造 RowData
          out.collect(rowData);
      }
   }
   

2. 注册自定义反序列化器
   在 Flink 作业中注册自定义反序列化器：

   PostgreSQLSource.<RowData>builder()
      .deserializer(new CustomDeserializer())
      .build();
   

重要提醒

• 时区一致性：确保 PostgreSQL、Flink 和下游系统（如 Hologres）的时区配置保持一致，避免因时区差异导致数据偏差。
• 数据迁移风险：如果修改 PostgreSQL 表结构或数据，请务必在测试环境中验证，避免对生产环境造成影响。
• 版本兼容性：上述解决方案适用于实时计算引擎 VVR 8.0.6 及以上版本。如果使用较低版本，请升级到最新版本以获得更好的兼容性和功能支持。

通过以上方法，您可以有效解决 Flink CDC 同步 PostgreSQL 数据时时间偏移 8 小时的问题。