Flink: преобразование binlog в несколько DTO и метод преобразования во flink

Новичок в Kafka, Flink и Tidb. Предположим, у меня есть три исходные таблицы MySql s_a, s_b и s_c, и я хочу собирать записи для целевой таблицы TiDb t_a и t_b в реальном времени. Правила сопоставления

`s_a`  --> `t_a`                   
`s_b` union `s_c`   ---> `t_b`  with some transformation (e.g., field remapping).

Решение, которое я принял, — это kafka + Flink с приемником Tidb, где изменения binlog подписываются на тему Kafka; Flink использует тему и записывает преобразованный результат в Tidb. Проблема для меня в части кода flink:

1. как я могу легко восстановить строку json (содержащую информацию об операциях, таблицах), опрашиваемую из kafka, для различных операций DTO (например, вставка/создание t_a или t_b ). Я нашел инструмент под названием Debezium в качестве коннектора Kafka&Flink, но похоже, что он требует равенства между исходной и целевой таблицами.

2. Как написать преобразование VKDataMapper, если у меня есть несколько целевых таблиц. Мне трудно определить T, так как это может быть t_a DTO (объект передачи данных) или t_b DTO.

Существующий пример кода для меня выглядит так:

// Основная процедура.

   StreamExecutionEnvironment environment =
                StreamExecutionEnvironment.getExecutionEnvironment();
       //consume is FlinkkafkaConsumer. TopicFilter returns true. 
        environment.addSource(consumer).filter(new TopicFilter()).map(new VKDataMapper())
                .addSink(new TidbSink());

        try {
            environment.execute();
        } catch (Exception e) {
            log.error("exception {}", e);
        }
 
 
 public class VKDataMapper implements MapFunction<String, T> {

    @Override
    public T map(String value) throws Exception {
        //How T can represents both `T_a data DTO` `T_b`...., 
        return null;
    }

}