SQL如何实现多表同步

[xiaoyu19950911]

假设mysql单库下面有100张不同的业务表（非分表）

1. 如果要使用sql同步50张表，是否需要写50个sql来实现，这样会订阅同一个mysql库50次，这样成本比较高而且会对mysql产生压力，有没有什么好的办法吗？
2. 如果50张表可以通过一个sql只订阅一次，那么当该库中新增需要同步的表时，如何将该表纳入订阅？新增表目前看下来只能重新新增订阅任务
   @wuchong

[kewant]

Refer to the official documentation to use regular expressions to match multiple libraries and multiple tables, should be able to solve your problem？https://ververica.github.io/flink-cdc-connectors/release-1.4/content/connectors/mysql-cdc.html#connector-options

[xiaoyu19950911]

But not all tables are regular and these tables have different columns

[kewant]

I'm not sure if I can set null for columns that are present in A but not in B ,Now it seems that only the same columns in both tables can be matched

[xiaoyu19950911]

I tested it and confirmed that it would be null when the column that are present in A but not in B,but if we have 100 tables,we can not define columns for all tables in a ddl

[zjchs66]

有好的解决方案吗？

[shadon178]

SQL用的少，但是可以用java api去实现呀。

[hiliuxg]

假设mysql单库下面有100张不同的业务表（非分表）

1. 如果要使用sql同步50张表，是否需要写50个sql来实现，这样会订阅同一个mysql库50次，这样成本比较高而且会对mysql产生压力，有没有什么好的办法吗？
2. 如果50张表可以通过一个sql只订阅一次，那么当该库中新增需要同步的表时，如何将该表纳入订阅？新增表目前看下来只能重新新增订阅任务
   @wuchong

我们自己实现了一个方案tablesource，基于flink cdc 2.0 api 实现整库同步到kafka，然后下游又实现了一个format，选择固定的表，直接同步

[dxs123]

下游format是什么意思？可否共享一下你的方案

[caesar168]

2.2版本说mysql已经支持动态添加新表，不用再起新作业了。但是不知道其他数据库支持不支持。

[lykpass]

flink2.xz怎么获取表结构信息呢，canal 监听的log中有schema但是cdc api中获取不到schema

[wangxiangyu]

我们目前实现了一个方案，做法是
在mysql cdc sql connector中，去ddl，然后适配所有的debezium record，对于所有库和表的全部字段全部统一序列化为json，然后下游在根据元信息去（库、表）分流

[dxs123]

下游分流的话那不就还是多少个表多少个流吗？怎么样能做到一个库仅占用一个流呢

[juyonychen]

flink cdc 用于监听mysql库，是支持配置多个表的（正则或者枚举的方式）。如果你要实现SQL的方式直接用SQL client不知道咋实现，但是APi方式的话：stmtSet.addInsertSql("insert_02")；stmtSet.addInsertSql("insert_01")。

现在在使用flink cdc的时候最大的问题是：我们无法监听到上游mysql的字段变更，这会导致很多问题，例如上游字段增加了字段后，我们实时同步数据到SR、hudi时，新增的字段是没办法实时同步到下游的，即使新增字段传递到了下游，也会因为SR、hudi等schema信息是固定的导致数据依旧会丢失。 如果这部分数据不想丢失的话，目前flink-cdc只能任务重启重新全量同步（不依赖CP)。

以上会导致：如果一个job监听的表过多的话，在重启flink任务时，需要的表原本不需要全量同步，现在因为一张或多张表的schema信息变更后也被迫需要全量同步。 如果把mysql中的表拆分成一个表一个job，这又会导致启动监听的binlog任务过多，增加网络和不必要的处理开销。更加不是一个好的方案。

所以：期望flink cdc每次重启job时，不仅仅是可以动态增加表还可以支持一下功能：
1、可以指定哪些表时需要全量同步。不在列表范围内的按照增量同步；
2、和动态增加表功能恰恰相反，可以动态的删除表。

目前来说的话，不解决以上的问题，flink-cdc只能用于一些schema变更非常不频繁的业务场景下。 对于schema几乎每周都在变化的场景，可能就不太适合了。毕竟每次全量数据同步的开销确实都非常的大。

[lykpass]

现在通过api是能获取到表结构变更的，如果还能获取表结构信息就好了，自己实现sink的逻辑，就能解决这些问题

[GourdErwa]

API 方式可以分流，多表一个源。
可参考 MysqlCDC2ChangeLog

[GourdErwa]

API 方式可以分流，多表一个源。 可参考 MysqlCDC2ChangeLog

如果是多表关联 形成一张宽表可以实现吗

可以呀，按 sql 写就行。

[J-Orange]

写sql就不能进行数据处理了

[GourdErwa]

写sql就不能进行数据处理了

sql 和 API 互通的，现在没明白你都需求了。

[J-Orange]

写sql就不能进行数据处理了

sql 和 API 互通的，现在没明白你都需求了。

怎么互通啊，sql多表关联只需要join多个多表数据源就可以了，stream要怎么操作？

[GourdErwa]

写sql就不能进行数据处理了

sql 和 API 互通的，现在没明白你都需求了。

怎么互通啊，sql多表关联只需要join多个多表数据源就可以了，stream要怎么操作？

stream 最后也是注册成 table了，table.join(table) 操作