直接使用Flink CDC SQL的写法，一个Job只能同步一个表的数据，至于原因，在此不再赘述。 直接上代码吧 将SourceRecord类转化为自定义的JsonRecord类型 public class JsonStringDebeziumDeserializationSchema implements DebeziumDeserializationSchema<JsonRecord> { @Override public void deserialize(SourceRecord record, Collector<JsonRecord> out) throws Exception { Envelope.Operation op = Envelope.operationFor(record); Struct value = (Struct) record.value(); Schema valueSchema = record.valueSchema(); String tableName = record.topic(); //out.collect("source...
