Realtime Compute for Apache Flink:Problèmes de synchronisation des données

Modifications du schéma JSON de Kafka vers Hologres

La solution classique consiste à arrêter le déploiement Flink, modifier le code et le schéma de la table Hologres, puis redéployer. Toutefois, ce processus peut entraîner des retards dans le traitement des données et introduire des erreurs inutiles.

Realtime Compute for Apache Flink propose les optimisations suivantes pour résoudre ce problème :

• Évolution automatique du schéma pour le format JSON Kafka : si le schéma JSON change, le système synchronise automatiquement le nouveau schéma vers Hologres sans qu'il soit nécessaire d'arrêter le déploiement ou de modifier les instructions SQL.

• Inférence de type pour le format JSON Kafka : il n'est pas nécessaire de déclarer les types de données des champs JSON dans la DDL.

• Expansion récursive des champs JSON imbriqués : par exemple, pour les données JSON {"nested": {"col": true}}, le champ col est développé en nested.col.

Erreur : Encountered change event for table xxx.xxx whose schema isn't known to this connector

• Problème

  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

• Cause

  Lorsque vous utilisez la fonctionnalité CDAS ou CTAS pour lire de nouvelles tables, cette erreur peut se produire pour les raisons suivantes :

  • Le compte de base de données peut ne pas disposer des permissions requises sur certaines bases de données utilisées par le déploiement.

  • Le déploiement est configuré avec 'debezium.snapshot.mode'='never'. Ce paramètre oblige le connecteur à lire depuis le début du journal binaire (binlog). Si le schéma de table associé à un événement de modification précoce dans le journal binaire ne correspond pas au schéma de table actuel, cette erreur se produit.

  • Cette erreur peut survenir si Debezium rencontre une modification qu'il ne peut pas analyser, telle que DEFAULT (now()).

• Solution

  • Assurez-vous que le compte de base de données dispose de toutes les permissions de base de données requises par le déploiement. Ces permissions incluent généralement la lecture, l'écriture et la modification des données, ainsi que la création et la modification des schémas et des tables. Pour plus d'informations, consultez la documentation du connecteur correspondante.

  • Utilisez le paramètre 'debezium.inconsistent.schema.handling.mode' = 'warn' pour éviter les erreurs. N'utilisez pas 'debezium.snapshot.mode'='never'.

  • Continuez à surveiller les journaux. Recherchez les messages io.debezium.connector.mysql.MySqlSchema WARN, qui fournissent des détails sur les modifications spécifiques qui ne peuvent pas être analysées. Par exemple, Debezium ne peut pas analyser DEFAULT (now()).

Erreur : Currently does not support merge StreamExecMiniBatchAssigner type ExecNode in CTAS/CDAS syntax

• Détails

  Une erreur se produit lors du déploiement ou du démarrage d'un déploiement :

  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)

• Cause

  Les déploiements CTAS et CDAS ne prennent pas en charge les opérations MiniBatch. Toutefois, la configuration table.exec.mini-batch.enabled: 'true' est activée dans le déploiement SQL.

• Solution

  Supprimez les configurations liées à MiniBatch du déploiement SQL.

  • Pour les déploiements ayant échoué

    Dans la section Other Configuration de l'onglet Deployment Defaults de la page Configurations, supprimez les informations de configuration liées à MiniBatch ou définissez table.exec.mini-batch.enabled sur false. Après avoir enregistré les modifications, recréez et déployez le déploiement SQL.

  • Pour les déploiements existants

    Si un déploiement a ignoré la vérification approfondie avant le déploiement et est terminé, accédez à l'onglet Deployment Details de la page Deployments. Dans la section Other Configuration des Parameters, supprimez les informations de configuration MiniBatch ou définissez table.exec.mini-batch.enabled sur false, puis redémarrez le déploiement.