Flink CDC 中mysql timestamp全量读取的时间正常，增量阶段少8小时，有人遇到？

在使用 Flink CDC 读取 MySQL 数据时，如果全量阶段的时间戳正常，但在增量阶段时间戳少了 8 小时，这通常是由于时区配置不一致导致的。以下是详细的原因分析和解决方案：

问题原因

1. server-time-zone 参数未正确配置
   在解析 Binlog 数据中的 timestamp 字段时，Flink CDC 使用 server-time-zone 参数来指定 MySQL 服务器的时区。如果该参数未与 MySQL 服务器的实际时区一致，可能会导致时间戳解析错误。

   • 默认情况下，Flink CDC 的 server-time-zone 参数可能被设置为 UTC+8（Asia/Shanghai），但如果 MySQL 服务器实际运行在其他时区（如 UTC），则会导致增量阶段的时间戳少 8 小时。

2. 自定义序列化器问题
   如果您在 DataStream 中使用了自定义序列化器（例如实现了 DebeziumDeserializationSchema 接口），并且在解析 timestamp 类型数据时未正确处理时区信息，也可能导致时间戳偏差。

3. TIMESTAMP 和 TIMESTAMP_LTZ 的差异
   Flink 中的 TIMESTAMP 类型是无时区的时间戳，而 TIMESTAMP_LTZ 是带有时区的时间戳。如果下游系统（如 Hologres）将 TIMESTAMP_LTZ 当作无时区时间处理，则可能导致时间戳偏差。

解决方案

1. 配置正确的 server-time-zone 参数

确保在 Flink CDC 的 MySQL 源表配置中，server-time-zone 参数与 MySQL 服务器的实际时区一致。例如：

CREATE TABLE mysql_source (
    id INT,
    event_time TIMESTAMP(3)
) WITH (
    'connector' = 'mysql-cdc',
    'hostname' = 'your-mysql-host',
    'port' = '3306',
    'username' = 'your-username',
    'password' = 'your-password',
    'database-name' = 'your-database',
    'table-name' = 'your-table',
    'server-time-zone' = 'UTC' -- 根据 MySQL 实际时区调整
);

• 如果 MySQL 服务器运行在 UTC 时区，请将 server-time-zone 设置为 'UTC'。
• 如果 MySQL 服务器运行在 Asia/Shanghai 时区，请将 server-time-zone 设置为 'Asia/Shanghai'。

2. 检查自定义序列化器

如果您在 DataStream 中使用了自定义序列化器，请参考 RowDataDebeziumDeserializeSchema 中对 timestamp 类型的解析逻辑，并确保在解析时提供正确的时区信息。例如：

private TimestampData convertToTimestamp(Object dbzObj, Schema schema) {
    if (dbzObj instanceof Long) {
        switch (schema.name()) {
            case Timestamp.SCHEMA_NAME:
                return TimestampData.fromEpochMillis((Long) dbzObj);
            case MicroTimestamp.SCHEMA_NAME:
                long micro = (long) dbzObj;
                return TimestampData.fromEpochMillis(micro / 1000, (int) (micro % 1000 * 1000));
            case NanoTimestamp.SCHEMA_NAME:
                long nano = (long) dbzObj;
                return TimestampData.fromEpochMillis(nano / 1000_000, (int) (nano % 1000_000));
        }
    }
    LocalDateTime localDateTime = TemporalConversions.toLocalDateTime(dbzObj, serverTimeZone);
    return TimestampData.fromLocalDateTime(localDateTime);
}

确保 serverTimeZone 参数与 MySQL 服务器的时区一致。

3. 检查下游系统的时区处理

如果下游系统（如 Hologres）将 TIMESTAMP_LTZ 当作无时区时间处理，可能会导致时间戳偏差。建议在写入下游系统时，显式转换时间戳类型。例如：

TO_TIMESTAMP_TZ(event_time, 'Asia/Shanghai')

通过 TO_TIMESTAMP_TZ 函数，可以将时间戳根据指定时区进行转换。

4. 确保全量和增量阶段的时区一致性

在全量阶段，Flink CDC 直接从 MySQL 表中读取数据，通常不会受到 server-time-zone 参数的影响。但在增量阶段，Flink CDC 解析 Binlog 数据时会依赖 server-time-zone 参数。因此，确保全量和增量阶段的时区配置一致非常重要。

重要提醒

• 检查 MySQL 服务器的时区设置：可以通过以下 SQL 查询 MySQL 服务器的时区配置：

  SHOW VARIABLES LIKE '%time_zone%';
  

  确保 system_time_zone 和 time_zone 的值与 Flink CDC 的 server-time-zone 参数一致。

• 升级 MySQL 版本：如果使用的是 RDS MySQL 5.6 只读实例，建议升级到更高版本或切换到可写实例，以避免 Binlog 文件简化导致的问题。

通过以上步骤，您可以解决 Flink CDC 在增量阶段时间戳少 8 小时的问题。如果问题仍然存在，请进一步检查 MySQL 服务器的时区配置以及下游系统的时区处理逻辑。