1. 程式人生 > 其它 >【Flink SQL】 MYSQL CDC 到 Upsert Kafka的時間戳使用ISO8601

【Flink SQL】 MYSQL CDC 到 Upsert Kafka的時間戳使用ISO8601

需求

需要將Kafka的資料匯入到Apache Druid,而Apache Druid的Ingestion時區是預設UTC+0的。

不能使用如下時間格式:

  • '2021-09-28 10:00:00', 會被當成'2021-09-28T10:00:00Z'。

總結

  1. MYSQL CDC source 必須是timestamp,不能是timestamp_ltz,否則Flink作業啟動後,CDC的connector報錯。
  2. MYSQL CDC source server-time-zone 加了UTC+8,報錯(說是找不到UTC+08:00這個TimeZone Entry),只能加 'server-time-zone' = 'GMT+8'
  3. Upsert Kafka 要加 ISO-8610 屬性,但是加了會把當前時間作為UTC+0,不加直接變成原始SQL時間戳。
  4. Upsert Kafka 時間列可以設定為 TIMESTAMP_LTZ,否則JSON依舊變成UTC+0。
  5. KAFKA Connector 屬性叫json.timestamp-format.standard,KAFKA UPSERT Connector 的屬性叫value.json.timestamp-format.standard(slankka注) 。

Kafka Upsert Connector:

  'value.json.timestamp-format.standard' = 'ISO-8601',
  'key.format' = 'json',
  'value.format' = 'json',
  'value.fields-include' = 'ALL' -- EXCEPT_KEY:key不出現kafka訊息的value中, 預設ALL(slankka注)

附錄

如果CDC Source的列定義為timestamp_ltz,則報錯如下: