Flinkcdc如何指定表的隐藏列为主键啊？例如MySQL中的_rowid？

展开

收起

cuicuicuic 2023-12-10 20:55:30 48 0

3 条回答

写回答

取消提交回答

小周sir

面对过去，不要迷离；面对未来，不必彷徨；活在今天，你只要把自己完全展示给别人看。

2000元阿里云代金券免费领取，2核4G云服务器仅664元/3年，新老用户都有优惠，立即抢购>>>

在 Flink CDC 中，可以通过指定表的隐藏列为主键来确保数据的完整性和一致性。对于 MySQL 中的 _rowid 列，可以按照以下步骤进行操作：

首先，需要创建一个 MySQL 连接池，并将其注册为一个 TableEnvironment 对象的属性。例如：

StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
TableEnvironment tableEnv = TableEnvironment.create(env);
String url = "jdbc:mysql://localhost:3306/mydatabase";
String user = "root";
String password = "password";
String driverName = "com.mysql.jdbc.Driver";
Properties properties = new Properties();
properties.setProperty("user", user);
properties.setProperty("password", password);
properties.setProperty("driverName", driverName);
BridgeSourceFunction<Row> sourceFunction = MySqlSource.<Row>builder()
        .hostname("localhost")
        .port(3306)
        .databaseList(Arrays.asList("mydatabase"))
        .tableList(Arrays.asList("mytable"))
        .username(user)
        .password(password)
        .deserializer(new RowDebeziumDeserializationSchema())
        .build();
tableEnv.registerTableSource("mysource", sourceFunction);

然后，可以使用 SQL 语句创建一个新的表，并将 _rowid 列指定为主键。例如：

String createTableSQL = "CREATE TABLE mytable_new (id BIGINT NOT NULL, name VARCHAR(50), age INT, PRIMARY KEY (id)) ENGINE=InnoDB";
tableEnv.executeSql(createTableSQL);

在这个例子中，mytable_new 是新创建的表名，id 是主键列名。可以根据实际需求修改表名和列名。

最后，可以使用 Flink SQL 将原始表中的数据插入到新表中，并使用 _rowid 列作为主键。例如：

String insertSQL = "INSERT INTO mytable_new (id, name, age) SELECT rowid, name, age FROM mysource";
tableEnv.executeSql(insertSQL);

在这个例子中，mysource 是原始表的名称。可以根据实际需求修改表名。

2023-12-11 16:23:37

赞同展开评论打赏

xin在这

2000元阿里云代金券免费领取，2核4G云服务器仅664元/3年，新老用户都有优惠，立即抢购>>>
https://forums.mysql.com/read.php?61,368131,379277 ，此回答整理自钉群“Flink CDC 社区”

2023-12-11 11:58:09

赞同展开评论打赏
柒号华仔

2000元阿里云代金券免费领取，2核4G云服务器仅664元/3年，新老用户都有优惠，立即抢购>>>
在Flink CDC中，可以通过指定表的隐藏列为主键来实现。隐藏列是指在表中存在但不会被显示的列，例如MySQL中的"_rowid"列。

要指定隐藏列为主键，可以按照以下步骤进行操作：

在创建CDC Source时，指定表的主键列。可以使用JdbcSourceBuilder类的setPrimaryKeyFields()方法来指定主键列。例如：
JdbcSource source = JdbcSource.builder()
.setDrivername("com.mysql.cj.jdbc.Driver")
.setDBUrl("jdbc:mysql://localhost:3306/mydatabase")
.setUsername("username")
.setPassword("password")
.setTable("mytable")
.setPrimaryKeyFields("_rowid") // 指定隐藏列为主键
.setRowConverter(new MyRowConverter())
.build();
在RowConverter中定义将数据库行转换为Flink行的逻辑。在转换过程中，需要将隐藏列包含在输出的行中。例如：
public class MyRowConverter implements JdbcRowConverter {
@Override
public RowData convertRow(ResultSet resultSet, RowData reuse) throws Exception {
// 将隐藏列包含在输出的行中
String rowid = resultSet.getString("_rowid");
// 其他列的转换逻辑
// ...
return reuse;
}
}
通过以上步骤，你可以将表的隐藏列指定为主键，以便在Flink CDC中进行数据处理和操作。

2023-12-11 09:04:39

赞同展开评论打赏