【时区】Flink JDBC 和CDC时间字段时区 测试及时间基准

关联文章:
各种时间类型和timezone关系浅析

一、测试目的和值

1. 测试一般的数据库不含time zone的类型的时区。

• mysql timestamp(3) 类型
• postgres timestamp(3) 类型
• sqlserver datetime2(3) 类型
• oracle类型 TIMESTAMP(3) 类型
  在以下测试之中均为ts字段

2.测试CDC中元数据op_ts 时区

op_tsTIMESTAMP_LTZ(3) NOT NULL当前记录表在数据库中更新的时间。如果从表的快照而不是 binlog 读取记录，该值将始终为0。|
1

在以下测试中cdc表建表均使用ts_ms TIMESTAMP_LTZ(3) METADATA FROM 'op_ts' VIRTUAL 表示。
cdc在读取表时候分两个阶段：

1. 全量读取阶段，特点是jdbc读取，读取数据中op=r
2. 增量读取阶段，特点是log读取，读取数据中op=c或u或d
   op在截图中看到如3="r" 或者 3="r"，3是op字段的索引值。
   ts_ms在全量阶段读取数据以下成为READ数据
ts_ms在增量阶段读取数据以下成为CREATE数据

3. flink 数据时间表示和时区

flink Table中时间必须使用org.apache.flink.table.data.TimestampData对象表示。

@PublicEvolving  
public final class TimestampData implements Comparable<TimestampData> {  
    private final long millisecond;  
    private final int nanoOfMillisecond;
}
1
2
3
4
5

此类型使用如下两个值联合表示记录时间。并不记录时区数据。

实战测试：

@Test  
public void testTimeZone(){  
  
    // 常识：Epoch就是值utc的0时间点，是全局绝对时间点，本质是`ZoneOffset.of("+0")`下的0时间。与`January 1, 1970, 00:00:00 GMT`视为等同。  
    // GMT是前世界标准时，UTC是现世界标准时。UTC 比 GMT更精准，以原子时计时，适应现代社会的精确计时。  
    // 28800000=8*3600*1000。8小时毫秒值。  
  
    // 如下时间是+8时区的数据库存储的不带时区的时间：2023-09-28T09:43:20.320  
    long ts=1695894200320L;  
  
    // 如果将ts当做utc时间0时刻转为字符串则会导致时间+8 hour。2023-09-28 17:43:20。这是一般常用的在线转换时间的结果。因其默认是是epoch时间，所以转换后会+8h。  
    // 可见数据库读取的不带timezone时间的毫秒值，并不是以utc0时间(epoch)为基准的，而是以当前时区0为基准的。  
  
    // LocalDateTime对象本质支持LocalDate和LocalTime两个对象，LocalDate持有Integer的`年`,`月`,`日`。LocalTime则持有Integer的`时`,`分`,`秒`等和java.util.Date类型并不一样。  
    // LocalDateTime 的带有ZoneOffset方法比较难理解，此处：  
    // epochSecond 当然值的是epoch的秒数，是绝对时间概念和`java.util.Date.getTime()/1000`对应的，而offset是指此epoch秒数需要偏移的时间量。  
    // 内部代码是`long localSecond = epochSecond + offset.getTotalSeconds();`。  
  
    // 如下代码是正确的，因为java中的`java.util.Date`类和`java.sql.Timestamp`类型都是持有绝对时间的类，`Date.getTime`获得也是相对于Epoch的毫秒值（Returns the number of milliseconds since January 1, 1970, 00:00:00 GMT）。  
    LocalDateTime ldtFromDate = LocalDateTime.ofEpochSecond(new Date().getTime() / 1000, 0, ZoneOffset.of("+8"));  
    System.out.println(ldtFromDate);  // 2023-09-28T16:16:45。此时时钟也是16:17:44。  
    Date date0 = new Date(0); // number of milliseconds since the standard base time known as "the epoch"  
    System.out.println(date0.getTime()); // 0， date0.getTime()方法返回绝对时间Returns the number of milliseconds since January 1, 1970, 00:00:00 GMT  
  
    // 如下的提供`ZoneOffset.UTC`可以理解是告诉LocalDateTime我提供的epochSecond已是`localSecond=当地时间-当地时间的0点`不需要再做转换了。  
    LocalDateTime ldt0 = LocalDateTime.ofEpochSecond(0L, 0, ZoneOffset.UTC);  
    System.out.println(ldt0); // 1970-01-01T00:00  
    LocalDateTime ldt8 = LocalDateTime.ofEpochSecond(0L, 0, ZoneOffset.of("+8"));  
    System.out.println(ldt8); // 1970-01-01T08:00  
  
    // TimestampData 默认不会进行任何时区转换。也不存储任何时区信息。内部仅靠`long millisecond`和`int nanoOfMillisecond`存储信息，以便于序列化。  
    // millisecond 一般可以认为是本地时间。因其在toString方法中会不会进行时区转换，toString方法仅是调用了`toLocalDateTime()`，中进行简单运算，并最终调用`LocalDateTime.toString`方法。  
    TimestampData td0 = TimestampData.fromEpochMillis(0); // 相当于LocalDateTime.ofEpochSecond(0, 0, ZoneOffset.UTC)。  
    System.out.println(td0); // 1970-01-01T00:00。可见TimestampData输出转字符串的时间就是以utc时间为基准的这和java.util.Date类型是一致的。  
  
    LocalDateTime ldt = LocalDateTime.ofEpochSecond(  
            ts / 1000  
            , (int) (ts % 1000 * 1_000_000)  
            , ZoneOffset.UTC);  
    System.out.println(ldt); // 2023-09-28T09:43:20.320  
    TimestampData td = TimestampData.fromEpochMillis(ts);  
    System.out.println(td); // 2023-09-28T09:43:20.320  
  
    Date date = new Date(ts); // 注意：参数date(the specified number of milliseconds since the standard base time known as "the epoch")应该是epoch但此时ts并不是epoch基准的而是本地local基准的。  
    System.out.println(date); // Thu Sep 28 17:43:20 CST 2023,CST就是北京时间了，其在toString方法中`BaseCalendar.Date date = normalize();`进行了时区转换即+8了。  
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46

4. 测试组件版本

• flink 1.13
• flink-cdc 2.2.1
• flink-connector-jdbc 自己定制的，根据3.1.1-1.17版本修改而来。

二、本测试共测试四大数据库：

• mysql
• postgres
• sqlserver
• oracle

二、每种数据库测试8项：

• database-SQL
  直接从数据中读取数据，是测试的基准值
• cdc-RowData
  使用cdc的SQL API从数据库中读取值并在 com.ververica.cdc.debezium.table.AppendMetadataCollector#collect 方法中debug得到数据
• cdc-SQL(测试除ts_ms的字段)
  使用cdc的SQL API读取值使用flink sql-client查询，用于测试除ts_ms的字段。因ts_ms准确性需分两种情况讨论。
• cdc-SQL-RealTime(测试ts_ms)
  使用cdc的SQL API从读取值，左上角是系统时间，下侧是实时读取的数据。
• cdc-Read数据(测试snapshot读取ts_ms字段)
  测试snapshot读取ts_ms字段，即全量读取阶段的ts_ms值，按照flink-cdc官方解释此四个数据的全量阶段值均为0(1970-01-01 00:00:00)。非0即为不正确。
• cdc-Create数据(测试incremental读取ts_ms字段)
  测试incremental读取ts_ms字段，即增量读取阶段的ts_ms值。按照flink-cdc官方解释此四个数据的增量阶段值为数据日志记录时间。
• jdbc-RowData
  使用flink SQL API 读取connector是jdbc的表数据org.apache.flink.connector.jdbc.table.JdbcRowDataInputFormat#nextRecord的方法中debug得到数据。。不含tm_ms数据。
• jdbc-SQL
  使用flink SQL API 读取connector是jdbc的表数据。使用flink sql-client查询。。不含tm_ms数据。

三、测试过程数据

3.1 mysql

3.1.1 database-SQL

3.1.2 cdc-RowData

3.1.3 cdc-SQL(测试除ts_ms的字段)

3.1.4 cdc-SQL-RealTime(测试ts_ms)

如下：上侧（win系统显示时间截图），下侧（cdc-query的ts_ms）
如果基本一致（不是差值8h），说明cdc-query的ts_ms是正确的的。

3.1.5 cdc-Read数据(测试snapshot读取ts_ms字段)

3.1.6 cdc-Create数据(测试incremental读取ts_ms字段)

3.1.7 jdbc-RowData

3.1.8 jdbc-SQL

3.2 postgres

3.2.1 database-SQL

3.2.2 cdc

cdc-RowData

3.2.3 cdc-SQL(测试除ts_ms的字段)

3.2.4 cdc-SQL-RealTime(测试ts_ms)

3.2.5 cdc-Read数据(测试snapshot读取ts_ms字段)

3.2.6 cdc-Create数据(测试incremental读取ts_ms字段)

3.2.7 jdbc

jdbc-RowData

3.2.8 jdbc-SQL

3.3 sqlserver

3.3.1 database-SQL

3.3.2 cdc-RowData

3.3.3 cdc-SQL(测试除ts_ms的字段)

3.3.4 cdc-SQL-RealTime(测试ts_ms)

3.3.5 cdc-Read数据(测试snapshot读取ts_ms字段)

3.3.6 cdc-Create数据(测试incremental读取ts_ms字段)

3.3.7 jdbc-RowData

3.3.8 jdbc-SQL

3.4 oracle

3.4.1 database-SQL

3.4.2 cdc-RowData

3.4.3 cdc-SQL(测试除ts_ms的字段)

3.4.3 cdc-SQL-RealTime(测试ts_ms)

3.4.4 cdc-Read数据(测试snapshot读取ts_ms字段)

3.4.5 cdc-Create数据(测试incremental读取ts_ms字段)

3.4.7 jdbc-RowData

3.4.8 jdbc-SQL

四、结论

（1）数据库获取的without time zone在flink中都是以本地时间的存储的。可以使用LocalDateTime.ofEpochSecond(long epochSecond, int nanoOfSecond, ZoneOffset.UTC)直接获取。
（2）Flink中的TimestampData中存储的一般可以认为是本地时间。但需要注意：TimestampData 不可将 instant 相关方法和 localDateTime 、Timestamp 相关方法混用。因为instant代表与epoch时间差。而后两者代表与local是时间差。
（3）Flink程序中时间的标准值都是local本地的。因其在Sql API（sql-client）中打印出的结果会与原始数据库中打印的一致。

如下图中红色字体的是错误的数据，使用CDC需要额外注意并进行转换。

五、附录

5.1 查询数据库时区SQL

-- mysql 以：time_zone 为准，system_time_zone至服务器时区
show variables like '%time_zone%';

-- postgres
show time zone;

-- sqlserver
DECLARE
@TimeZone NVARCHAR(255)
EXEC
master.dbo.xp_instance_regread
N'HKEY_LOCAL_MACHINE'
,
N'SYSTEM\CurrentControlSet\Control\TimeZoneInformation'
,
N'TimeZoneKeyName'
,
@TimeZone
OUTPUT
SELECT
@TimeZone 

-- oracle
select dbtimezone from dual;

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25