Hologres：Flink/Blink即時消費Hologres Binlog

消費Hologres Binlog需要注意如下事項：

• 僅Hologres V0.9及以上版本支援消費Hologres Binlog；僅Hologres V1.3.21及以上版本支援配置引擎白名單，HologresV1.3.21以下版本當前暫不支援配置引擎白名單，開啟白名單後，會造成Binlog消費失敗。如果您的執行個體版本低於所要求執行個體版本，請您加入HologresDingTalk群進行反饋，詳情可參見線上支援。

• Hologres支援單表層級的Binlog功能，支援行存表和列存表，以及從Hologres V1.1版本開始支援行列共存表。開啟Binlog後，理論上列存表的開銷要大於行存表的開銷。因此對於資料更新頻繁的情境，建議為使用行存儲存格式的表開啟Binlog。

• Hologres Binlog的支援情況以及開啟、配置Hologres Binlog，請參見訂閱Hologres Binlog。

• 僅阿里雲Flink支援消費Hologres Binlog。Holohub模式下Flink消費Hologres Binlog只支援單一資料型別，從Flink 6.0.3版本開始，支援通過JDBC模式消費Hologres Binlog，相比Holohub，JDBC支援更多的資料類型，詳情請參見Blink/Flink與Hologres的資料類型映射。同時增加了部分許可權限制，詳情請參見許可權說明。

• 目前不支援消費分區父表的Binlog。

• Hologres V2.0版本起有限支援Holohub模式；V2.1版本起下線Holohub模式，全面轉為JDBC模式。在您升級Hologres版本前，請參考Holohub模式切換到JDBC模式，查看您當前正在使用Holohub模式的Flink任務，並按步驟升級Flink VVR作業版本，然後升級Hologres執行個體。

該模式下Source消費的Binlog資料是作為普通的Flink資料傳遞給下遊節點的，即所有資料都是作為Insert類型的資料，可以根據業務情況選擇如何處理特定hg_binlog_event_type類型的資料。Hologres表開啟Binlog後，在Flink中源表（非CDC模式）使用如下DDL可以即時消費Binlog。

create table test_message_src_binlog_table(
  hg_binlog_lsn BIGINT,
  hg_binlog_event_type BIGINT,
  hg_binlog_timestamp_us BIGINT,
  id INTEGER,
  title VARCHAR,
  body VARCHAR
) with (
  'connector'='hologres',
  'dbname'='<yourDbname>',
  'tablename'='<yourTablename>',
  'username'='<yourAccessID>',
  'password'='<yourAccessSecret>',
  'endpoint'='<yourEndpoint>',
  'binlog' = 'true',
  'binlogMaxRetryTimes' = '10',
  'binlogRetryIntervalMs' = '500',
  'binlogBatchReadSize' = '100'
);

• 三個binlogxxx 參數表示Binlog系統欄位，命名和類型是固定的不能修改。

• 其他欄位是跟使用者欄位一一對應，必須是全小寫。

該模式下Source消費的Binlog資料，將根據hg_binlog_event_type自動為每行資料設定準確的Flink RowKind類型（INSERT、DELETE、UPDATE_BEFORE、UPDATE_AFTER），這樣就能完成表的資料的鏡像同步，類似MySQL和Postgres的CDC功能。

Hologres表開啟Binlog後，在Flink中源表（CDC模式）使用如下DDL可以即時消費Binlog。

create table test_message_src_binlog_table(
  id INTEGER,
  title VARCHAR,
  body VARCHAR
) with (
  'connector'='hologres',
  'dbname'='<yourDbname>',//Hologres的DB名
  'tablename'='<yourTablename>',//Hologres的表名
  'username'='<yourAccessID>',//當前帳號的access id
  'password'='<yourAccessSecret>',//當前帳號的access key
  'endpoint'='<yourEndpoint>',//Hologres的vpc網路地址
  'binlog' = 'true',
  'cdcMode' = 'true',
  'binlogMaxRetryTimes' = '10',
  'binlogRetryIntervalMs' = '500',
  'binlogBatchReadSize' = '100'
);