Flink データ同期 FAQ（よくある質問）- Realtime Compute for Apache Flink - Alibaba Cloud - Realtime Compute for Apache Flink

このトピックでは、データ同期に関する一般的な問題とそのソリューションについて説明します。

Kafka から Hologres への JSON スキーマの変更
エラー: このコネクタにスキーマが認識されていないテーブル xxx.xxx の変更イベントが検出されました
エラー: 現在、 CTAS/CDAS 構文では StreamExecMiniBatchAssigner 型の ExecNode のマージをサポートしていません。

Kafka から Hologres への JSON スキーマの変更

従来のソリューションでは、Flink デプロイメントを停止し、コードと Hologres のテーブルスキーマを修正してから、再デプロイする必要がありました。しかし、このプロセスはデータ処理の遅延を引き起こし、不要なエラーを発生させる可能性があります。

Realtime Compute for Apache Flink は、この問題に対処するために、以下の最適化を提供します：

Kafka JSON フォーマットの自動スキーマ進化：JSON スキーマが変更された場合、システムはデプロイメントを停止したり SQL ステートメントを修正したりすることなく、新しいスキーマを自動的に Hologres に同期します。
Kafka JSON フォーマットの型推論：DDL で JSON フィールドのデータの型を宣言する必要はありません。
ネストされた JSON フィールドの再帰的な展開：例えば、JSON データ {"nested": {"col": true}} の場合、col フィールドは nested.col に展開されます。

エラー: このコネクタは、スキーマが不明なテーブル xxx.xxx の変更イベントを検出しました

問題

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 内の初期の変更イベントに関連付けられたテーブルスキーマが現在のテーブルスキーマと一致しない場合、このエラーが発生します。
- Debezium が DEFAULT (now()) のような解析できない変更に遭遇した場合に、このエラーが発生することがあります。
ソリューション
- データベースアカウントが、デプロイメントに必要なすべてのデータベース権限を持っていることを確認してください。これらの権限には通常、データの読み取り、書き込み、変更、およびテーブルスキーマとテーブルの作成と変更が含まれます。詳細については、対応するコネクタのドキュメントをご参照ください。
- エラーを防ぐために、'debezium.inconsistent.schema.handling.mode' = 'warn' パラメーターを使用してください。'debezium.snapshot.mode'='never' は使用しないでください。
- 引き続きログを監視してください。io.debezium.connector.mysql.MySqlSchema WARN メッセージを探します。これらのメッセージには、解析できない特定の変更に関する詳細が記載されています。例えば、Debezium は DEFAULT (now()) を解析できません。

エラー: 現在、CTAS/CDAS 構文では StreamExecMiniBatchAssigner 型 ExecNode のマージはサポートされていません

詳細

デプロイメントのデプロイ中または開始時にエラーが発生します：

USE CATALOG `test-holo`;
CREATE TABLE IF NOT EXISTS test_sync
AS TABLE `test-mysql`.`flink_test`.`test_sync`
/*+ OPTIONS('server-id'='9021-9030') */;
Problem: 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 デプロイメントは、ミニバッチ操作をサポートしていません。しかし、SQL デプロイメントで table.exec.mini-batch.enabled: 'true' 構成が有効になっています。
ソリューション

SQL デプロイメントからミニバッチ関連の構成を削除してください。
- 失敗したデプロイメントの場合
  
  [設定] ページの [デプロイメントのデフォルト] タブにある [その他の設定] セクションで、ミニバッチ関連の構成情報を削除するか、table.exec.mini-batch.enabled を false に設定します。変更を保存した後、SQL デプロイメントを再作成してデプロイしてください。
- 既存のデプロイメントの場合
  
  デプロイメントがデプロイメント前の詳細チェックをスキップして完了した場合、[デプロイメント] ページの [デプロイメントの詳細] タブに移動します。[パラメーター] の [その他の設定] セクションで、ミニバッチ構成情報を削除するか、table.exec.mini-batch.enabled を false に設定し、デプロイメントを再起動します。