Новичок в 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:
как я могу легко восстановить строку json (содержащую информацию об операциях, таблицах), опрашиваемую из kafka, для различных операций DTO (например, вставка/создание
t_a
илиt_b
). Я нашел инструмент под названиемDebezium
в качестве коннектора Kafka&Flink, но похоже, что он требует равенства между исходной и целевой таблицами.Как написать преобразование
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;
}
}