不用flinkcdc方式，通过flink将mysql的表还原到另一个表里该如何实现

如果不使用 flinkcdc 方式，可以通过 Flink 的 JDBC Connector 实现将 MySQL 表数据还原到另一个表中。

具体步骤如下：

1. 在 Flink 的 pom.xml 文件中添加 JDBC Connector 依赖：

<dependency>
    <groupId>org.apache.flink</groupId>
    <artifactId>flink-connector-jdbc_${scala.binary.version}</artifactId>
    <version>${flink.version}</version>
</dependency>

2. 编写 Flink 作业，使用 JDBC Connector 将 MySQL 表数据读取出来并写入到另一个表中，示例代码如下：

import org.apache.flink.api.java.ExecutionEnvironment;
import org.apache.flink.connector.jdbc.JdbcConnectionOptions;
import org.apache.flink.connector.jdbc.JdbcInputFormat;
import org.apache.flink.connector.jdbc.JdbcOutputFormat;
import org.apache.flink.table.api.EnvironmentSettings;
import org.apache.flink.table.api.TableEnvironment;
import org.apache.flink.table.api.bridge.java.BatchTableEnvironment;
import org.apache.flink.types.Row;

import java.sql.Types;

public class RestoreTableJob {
    public static void main(String[] args) throws Exception {
        // 创建执行环境
        ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
        BatchTableEnvironment tableEnv = TableEnvironment.create(EnvironmentSettings.newInstance().useBlinkPlanner().inBatchMode());

        // 读取 MySQL 表数据
        String query = "SELECT * FROM source_table";
        JdbcInputFormat jdbcInputFormat = JdbcInputFormat.buildJdbcInputFormat()
                .setDrivername("com.mysql.jdbc.Driver")
                .setDBUrl("jdbc:mysql://localhost:3306/test")
                .setUsername("root")
                .setPassword("123456")
                .setQuery(query)
                .setRowTypeInfo(Types.INT, Types.STRING, Types.DOUBLE)
                .finish();
        DataSet<Row> inputDataSet = env.createInput(jdbcInputFormat);

        // 写入到另一个表中
        String insertSql = "INSERT INTO target_table VALUES (?, ?, ?)";
        JdbcOutputFormat jdbcOutputFormat = JdbcOutputFormat.buildJdbcOutputFormat()
                .setDrivername("com.mysql.jdbc.Driver")
                .setDBUrl("jdbc:mysql://localhost:3306/test")
                .setUsername("root")
                .setPassword("123456")
                .setQuery(insertSql)
                .setSqlTypes(new int[]{Types.INT, Types.STRING, Types.DOUBLE})
                .finish();
        inputDataSet.output(jdbcOutputFormat);

        // 执行作业
        env.execute();
    }
}

需要注意的是，在 `JdbcInputFormat` 和 `JdbcOutputFormat` 中需要设置连接 MySQL 数据库的相关信息，如 `setDrivername`、`setDBUrl`、`setUsername` 和 `setPassword` 等，以及设置表中字段的类型，如 `setRowTypeInfo` 和 `setSqlTypes` 等。

此外，如果需要对数据进行转换或处理，可以在 Flink 作业中添加相应的转换算子，比如 `map`、`filter` 和 `reduce` 等。