Realtime Compute for Apache Flink:Masalah sinkronisasi data

Perubahan skema JSON dari Kafka ke Hologres

Solusi konvensional adalah menghentikan penerapan Flink, memodifikasi kode dan skema tabel Hologres, lalu menerapkannya kembali. Namun, proses ini dapat menyebabkan penundaan pemrosesan data dan memperkenalkan error yang tidak perlu.

Realtime Compute for Apache Flink menyediakan optimasi berikut untuk mengatasi masalah ini:

• Evolusi skema otomatis untuk format JSON Kafka: Jika skema JSON berubah, sistem secara otomatis menyinkronkan skema baru ke Hologres tanpa perlu menghentikan penerapan atau memodifikasi pernyataan SQL.

• Inferensi tipe untuk format JSON Kafka: Anda tidak perlu mendeklarasikan tipe data bidang JSON dalam DDL.

• Ekspansi rekursif bidang JSON bersarang: Misalnya, untuk data JSON {"nested": {"col": true}}, bidang col diekspansi menjadi nested.col.

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

• Permasalahan

  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

• Penyebab

  Saat Anda menggunakan fitur CDAS atau CTAS untuk membaca tabel baru, error ini dapat terjadi karena alasan berikut:

  • Akun database mungkin tidak memiliki izin yang diperlukan pada beberapa database yang digunakan oleh penerapan tersebut.

  • Penerapan dikonfigurasi dengan 'debezium.snapshot.mode'='never'. Pengaturan ini menyebabkan konektor membaca dari awal binlog. Jika skema tabel yang terkait dengan event perubahan awal dalam binlog tidak sesuai dengan skema tabel saat ini, error ini akan muncul.

  • Error ini juga dapat terjadi jika Debezium menemukan perubahan yang tidak dapat diurai, seperti DEFAULT (now()).

• Solusi

  • Pastikan akun database memiliki semua izin yang dibutuhkan oleh penerapan tersebut. Izin ini biasanya mencakup membaca, menulis, dan memodifikasi data, serta membuat dan memodifikasi skema tabel dan tabel itu sendiri. Untuk informasi lebih lanjut, lihat dokumentasi konektor terkait.

  • Gunakan parameter 'debezium.inconsistent.schema.handling.mode' = 'warn' untuk mencegah error. Hindari penggunaan 'debezium.snapshot.mode'='never'.

  • Lanjutkan pemantauan log. Cari pesan io.debezium.connector.mysql.MySqlSchema WARN, yang memberikan detail tentang perubahan spesifik yang tidak dapat diurai. Misalnya, Debezium tidak dapat mengurai DEFAULT (now()).

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

• Detail

  Terjadi error saat menerapkan atau memulai penerapan:

  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)

• Penyebab

  Penerapan CTAS dan CDAS tidak mendukung operasi MiniBatch. Namun, konfigurasi table.exec.mini-batch.enabled: 'true' diaktifkan dalam penerapan SQL.

• Solusi

  Hapus konfigurasi terkait MiniBatch dari penerapan SQL.

  • Untuk penerapan yang gagal

    Pada bagian Other Configuration di tab Deployment Defaults pada halaman Configurations, hapus informasi konfigurasi terkait MiniBatch atau atur table.exec.mini-batch.enabled menjadi false. Setelah menyimpan perubahan, buat ulang dan terapkan penerapan SQL tersebut.

  • Untuk penerapan yang sudah ada

    Jika suatu penerapan melewati pemeriksaan pra-penerapan mendalam dan telah selesai, buka tab Deployment Details pada halaman Deployments. Pada bagian Other Configuration di Parameters, hapus informasi konfigurasi MiniBatch atau atur table.exec.mini-batch.enabled menjadi false, lalu restart penerapan tersebut.