全部產品
Search

搜尋本產品

    Realtime Compute for Apache Flink：資料同步問題

    文件中心

    Realtime Compute for Apache Flink：資料同步問題

    更新時間：Jun 08, 2026

    本文為您介紹資料同步方面的常見問題。

    同步Kafka JSON資料到Hologres，作業運行過程中JSON Schema發生改變，如何解決？

    常規做法是停止Flink作業，修改代碼並調整Hologres表結構後重新上線，這樣可能會導致作業資料延時以及出現不必要的錯誤。

    針對以上的問題，Realtime ComputeFlink版做了以下最佳化來規避該問題：

    • 支援Kafka JSON format Schema自適應變更。即使JSON Schema發生改變，也無需停止作業或修改SQL語句，系統會自動將變更的Schema同步到Hologres。

    • 支援Kafka JSON format的類型推導，不需要在DDL語句聲明JSON裡面的欄位類型。

    • 支援遞迴式展開JSON的嵌套列，例如對於如下JSON資料：{"nested": {"col": true}}列col會被展開為nested.col。

    報錯：Encountered change event for table xxx.xxx whose schema isn't known to this connector

    • 問題詳情

      202x-xx-xx xx:xx:xx,xxx ERROR io.debezium.connector.mysql.BinlogReader    [] - Encountered change event 'Event{header=EventHeaderV4{timestamp=xxx, eventType=TABLE_MAP, serverId=xxx, headerLength=xxx, dataLength=xxx, nextPosition=xxx, flags=xxx}, data=TableMapEventData{tableId=xxx, database='xxx', table='xxx',
columnTypes=xxx, xxx..., columnMetadata=xxx,xxx,..., columnNullability={xxx,xxx,...}, eventMetadata=null}}' at offset {ts_sec=xxx, file=mysql-bin.xxx, pos=xxx,
gtids=xxx, ...server_id=xxx, event=xxx} for table xxx.xxx whose schema isn't known to this connector. One possible cause is an incomplete database history topic. Take
a new snapshot in this case.
Use the mysqlbinlog tool to view the problematic event: mysqlbinlog --start-position=30946 --stop-position=31028 --verbose mysql-bin.004419
202x-xx-xx xx:xx:xx,xxx ERROR io.debezium.connector.mysql.BinlogReader    [] - Error during binlog processing. Last offset stored = null, binlog
reader near position = mysql-bin.xxx/xxx
202x-xx-xx xx:xx:xx,xxx ERROR io.debezium.connector.mysql.BinlogReader    [] - Failed due to error: Error processing binlog event
org.apache.kafka.connect.errors.ConnectException: Encountered change event for table statistic.apk info whose schema isn't known to this connector
	at io.debezium.connector.mysql.AbstractReader.wrap(AbstractReader.java:241)
	~[veverica-connector-mysql-1.12-vvr-3.0.0-SNAPSHOT-jar-with-dependencies.jar:1.12-vvr-3.0.0-SNAPSHOT]
	at io.debezium.connector.mysql.AbstractReader.failed(AbstractReader.java:210)
	~[veverica-connector-mysql-1.12-vvr-3.0.0-SNAPSHOT-jar-with-dependencies.jar:1.12-vvr-3.0.0-SNAPSHOT]
	at io.debezium.connector.mysql.BinlogReader.handleEvent(BinlogReader.java:607)
	~[veverica-connector-mysql-1.12-vvr-3.0.0-SNAPSHOT-jar-with-dependencies.jar:1.12-vvr-3.0.0-SNAPSHOT]
	at com.github.shyiko.mysql.binlog.BinaryLogClient.notifyEventListeners(BinaryLogClient.java:1104)
	[veverica-connector-mysql-1.12-vvr-3.0.0-SNAPSHOT-jar-with-dependencies.jar:1.12-vvr-3.0.0-SNAPSHOT]
	at com.github.shyiko.mysql.binlog.BinaryLogClient.listenForEventPackets(BinaryLogClient.java:955)
	[veverica-connector-mysql-1.12-vvr-3.0.0-SNAPSHOT-jar-with-dependencies.jar:1.12-vvr-3.0.0-SNAPSHOT]
	at com.github.shyiko.mysql.binlog.BinaryLogClient.connect(BinaryLogClient.java:595)
	[veverica-connector-mysql-1.12-vvr-3.0.0-SNAPSHOT-jar-with-dependencies.jar:1.12-vvr-3.0.0-SNAPSHOT]
	at com.github.shyiko.mysql.binlog.BinaryLogClientIMT.run(BinaryLogClient.java:839)
	[veverica-connector-mysql-1.12-vvr-3.0.0-SNAPSHOT-jar-with-dependencies.jar:1.12-vvr-3.0.0-SNAPSHOT]
	at java.lang.Thread.run(Thread.java:834) [?:1.8.0_102]
Caused by: org.apache.kafka.connect.errors.ConnectException: Encountered change event for table xxx.xxx whose schema isn't known to this connector
	at io.debezium.connector.mysql.BinlogReader.informAboutUnknownTableIfRequired(BinlogReader.java:875)
	~[veverica-connector-mysql-1.12-vvr-3.0.0-SNAPSHOT-jar-with-dependencies.jar:1.12-vvr-3.0.0-SNAPSHOT]
	at io.debezium.connector.mysql.BinlogReader.handleUpdateTableMetadata(BinlogReader.java:849)
	~[veverica-connector-mysql-1.12-vvr-3.0.0-SNAPSHOT-jar-with-dependencies.jar:1.12-vvr-3.0.0-SNAPSHOT]
	at io.debezium.connector.mysql.BinlogReader.handleEvent(BinlogReader.java:590)
	~[veverica-connector-mysql-1.12-vvr-3.0.0-SNAPSHOT-jar-with-dependencies.jar:1.12-vvr-3.0.0-SNAPSHOT]
	... 5 more

    • 問題原因

      在開啟CDAS或CTAS新增表讀取功能時，出現該報錯的原因如下：

      • 當資料庫使用者缺少作業中使用的某些資料庫所需的許可權時，可能會出現該錯誤。

      • 作業中配置了'debezium.snapshot.mode'='never'。因為這會導致從Binlog開頭讀取資料，但是Binlog開頭的變更事件對應的Table Schema和當前表的Schema可能不匹配，所以會報該錯。

      • 遇到Debezium解析不了的變更，例如Debezium無法解析`DEFAULT (now())`，可能出現該錯誤。

    • 解決方案

      • 確保資料庫使用者擁有作業所需的所有資料庫許可權（通常需要讀取、寫入、修改資料、建立和修改表結構、以及建立表的許可權，具體要求的權限詳情請參見對應連接器文檔）。

      • 通過'debezium.inconsistent.schema.handling.mode' = 'warn'參數避免報錯，不建議使用'debezium.snapshot.mode'='never'，

      • 繼續觀察日誌，查詢io.debezium.connector.mysql.MySqlSchema WARN的log，會列印出具體無法解析的變更詳情，例如Debezium解析不了`DEFAULT (now())`。

    報錯：Currently does not support merge StreamExecMiniBatchAssigner type ExecNode in CTAS/CDAS syntax

    • 問題詳情

      在部署作業或者啟動作業時報錯：

      USE CATALOG `test-holo`;
CREATE TABLE IF NOT EXISTS test_tongbu
AS TABLE `test-mysql`.`flink_test`.`test_tongbu`
/*+ OPTIONS('server-id'='9021-9030') */;
問題: test-ctas
org.apache.flink.table.api.TableException: Currently does not support merge StreamExecMiniBatchAssigner type ExecNode in CTAS/CDAS syntax.
    at org.apache.flink.table.planner.plan.nodes.exec.processor.EvolvingTableSourceMergeProcessor$EvolvingNodeMergeVisitor.visit
(EvolvingTableSourceMergeProcessor.java:285)
    at org.apache.flink.table.planner.plan.nodes.exec.ExecNodeBase.accept(ExecNodeBase.java:249)
    at org.apache.flink.table.planner.plan.nodes.exec.visitor.ExecNodeVisitorImpl.lambda$visitInputs$0(ExecNodeVisitorImpl.java:34)
    at java.util.ArrayList.forEach(ArrayList.java:1249)
    at org.apache.flink.table.planner.plan.nodes.exec.visitor.ExecNodeVisitorImpl.visitInputs(ExecNodeVisitorImpl.java:34)
    at org.apache.flink.table.planner.plan.nodes.exec.visitor.ExecNodeVisitorImpl.visit(ExecNodeVisitorImpl.java:30)
    at org.apache.flink.table.planner.plan.nodes.exec.processor.EvolvingTableSourceMergeProcessor$EvolvingNodeMergeVisitor.visit
(EvolvingTableSourceMergeProcessor.java:263)
    at org.apache.flink.table.planner.plan.nodes.exec.ExecNodeBase.accept(ExecNodeBase.java:249)
    at org.apache.flink.table.planner.plan.nodes.exec.visitor.ExecNodeVisitorImpl.lambda$visitInputs$0(ExecNodeVisitorImpl.java:34)
    at java.util.ArrayList.forEach(ArrayList.java:1249)
    at org.apache.flink.table.planner.plan.nodes.exec.visitor.ExecNodeVisitorImpl.visitInputs(ExecNodeVisitorImpl.java:34)

    • 問題原因

      CTAS和CDAS作業不支援微批（MiniBatch）操作，但您的SQL流作業中卻啟用了table.exec.mini-batch.enabled: 'true'配置。

    • 解決方案

      在SQL流作業中去掉MiniBatch相關配置。

      • 未部署成功的SQL流作業

        在組態管理頁面的作業預設配置頁簽的其他配置中，刪除MiniBatch相關配置資訊或將table.exec.mini-batch.enabled取值設為false，儲存修改後重新建立並部署SQL流作業。

      • 已部署成功的SQL流作業

        如果作業已跳過部署前的深度檢查並完成部署，請前往作業營運頁面的部署詳情頁簽，在運行參數配置的其他配置中，刪除MiniBatch配置資訊或將table.exec.mini-batch.enabled取值設為false，並重啟作業。