有办法用flink-sql实现吗？

是的，您可以使用 Flink SQL 来实现 Flink CDC 的需求。以下是大致的步骤：

1. 定义 MySQL 数据源表：使用 CREATE TABLE 语句定义源表，并指定 CDC 连接器相关的配置参数，如主机名、端口号等。

CREATE TABLE source (
  ...
) WITH (
  'connector' = 'cdc',
  'hostname' = '...',
  'port' = '...',
  ...
)

2. 定义目标表：可以是 Flink 内部的其他表，也可以是外部系统（如 Kafka、Elasticsearch）。

CREATE TABLE sink (...) WITH (...);

3. 使用 Flink SQL 查询：编写查询语句从源表读取数据，并将结果写入目标表。

INSERT INTO sink
SELECT * FROM source;

您还可以在查询中执行数据转换或应用其他操作，例如投影、过滤、聚合等。

4. 提交 SQL 运行：通过编程 API 将 SQL 语句提交给 Flink 执行。

tableEnv.executeSql(sql);

这样，您就可以使用 Flink SQL 从 MySQL CDC 数据源读取数据并将其写入下游系统，实现 Flink CDC 的功能。

请注意，Flink SQL 支持常见的投影、过滤、聚合等操作，您可以根据需要自由组合来实现所需的数据同步逻辑。如果您需要更复杂的操作，您可能需要编写自定义的聚合函数

用 Flink SQL 实现 Flink CDC 需求完全可行。主要步骤是:
定义 MySQL 数据源表
sql
Copy
CREATE TABLE source (
...
) WITH (
'connector' = 'cdc',
'hostname' = '...',
'port' = '...',
...
)
定义目标表
可以是 Flink 内其他表,或者外部系统如 Kafka/ES。
sql
Copy
CREATE TABLE sink (...) WITH (...);
使用 Flink SQL 查询从源表读取数据,写入目标表
sql
Copy
INSERT INTO sink
SELECT FROM source
提交 SQL 运行
java
Copy
tableEnv.executeSql(sql);
这就实现了使用 Flink SQL 从 MySQL CDC 数据源读取数据,写入下游的功能。
你也可以在查询中实现数据转换:
sql
Copy
INSERT INTO sink
SELECT
id,
name,
amount 1.05
FROM source
Flink SQL 支持常用的:projection、filter、aggregation等,你可以自由组合实现需要的数据同步逻辑。
所以总的来说,完全可以用 Flink SQL 代替 Table API 实现 Flink CDC 任务。

自己写个自定义agg函数把，此回答整理自钉群“Flink CDC 社区”