flink读取mysql数据在插入到mysql 的flink代码，java写

以下是一个简单的 Flink 读取 MySQL 数据并将其插入到 MySQL 的示例代码。

首先，你需要在 pom.xml 文件中添加 MySQL 和 Flink Connector for MySQL 的依赖：

<dependencies>
    <dependency>
        <groupId>org.apache.flink</groupId>
        <artifactId>flink-java</artifactId>
        <version>1.11.2</version>
    </dependency>
    <dependency>
        <groupId>org.apache.flink</groupId>
        <artifactId>flink-streaming-java_2.12</artifactId>
        <version>1.11.2</version>
    </dependency>
    <dependency>
        <groupId>org.apache.flink</groupId>
        <artifactId>flink-connector-jdbc_2.12</artifactId>
        <version>1.11.2</version>
    </dependency>
    <dependency>
        <groupId>mysql</groupId>
        <artifactId>mysql-connector-java</artifactId>
        <version>8.0.23</version>
    </dependency>
</dependencies>

然后，可以使用以下代码读取 MySQL 数据并将其插入到另一个 MySQL 数据库中：

import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.functions.RuntimeContext;
import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaProducer;
import org.apache.flink.streaming.connectors.kafka.KafkaSerializationSchema;
import org.apache.flink.streaming.connectors.kafka.KafkaSink;
import org.apache.flink.streaming.connectors.kafka.KafkaTopicPartition;
import org.apache.flink.streaming.connectors.kafka.partitioner.FlinkKafkaPartitioner;
import org.apache.flink.streaming.util.serialization.JSONKeyValueDeserializationSchema;
import org.apache.flink.streaming.util.serialization.KeyedSerializationSchema;
import org.apache.flink.streaming.util.serialization.KeyedSerializationSchemaWrapper;
import org.apache.flink.types.Row;
import org.apache.flink.util.Preconditions;

import java.nio.charset.StandardCharsets;
import java.util.Optional;
import java.util.Properties;

public class FlinkMySQLExample {
    public static void main(String[] args) throws Exception {
        // set up the streaming execution environment
        final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        // configure properties for MySQL source
        Properties sourceProps = new Properties();
        sourceProps.setProperty("driverClassName", "com.mysql.cj.jdbc.Driver");
        sourceProps.setProperty("url", "jdbc:mysql://localhost:3306/mydatabase");
        sourceProps.setProperty("username", "myuser");
        sourceProps.setProperty("password", "mypassword");
        sourceProps.setProperty("query", "SELECT * FROM mytable");

        // create a MySQL source
        DataStream<Row> mysqlSource = env.createInput(new JDBCInputFormat()
                .setDBUrl(sourceProps.getProperty("url"))
                .setDrivername(sourceProps.getProperty("driverClassName"))
                .setUsername(sourceProps.getProperty("username"))
                .setPassword(sourceProps.getProperty("password"))
                .setQuery(sourceProps.getProperty("query"))
                .setRowTypeInfo(new RowTypeInfo(...)));

        // configure properties for MySQL sink
        Properties sinkProps = new Properties();
        sinkProps.setProperty("driverClassName", "com.mysql.cj.jdbc.Driver");
        sinkProps.setProperty("url", "jdbc:mysql://localhost:3306/mydatabase");
        sinkProps.setProperty("username", "myuser");
        sinkProps.setProperty("password", "mypassword");
        sinkProps.setProperty("table", "mytable");

        // create a MySQL sink
        mysqlSource.addSink(new JDBCOutputFormat()
                .setDBUrl(sinkProps.getProperty("url"))
                .setDrivername(sinkProps.getProperty("driverClassName"))
                .setUsername(sinkProps.getProperty("username"))
                .setPassword(sinkProps.getProperty("password"))
                .setQuery(String.format("INSERT INTO %s VALUES (?, ?, ...)", sinkProps.getProperty("table")))
                .setBatchInterval(1000)
                .setBatchSize(1000)
                .setSqlTypes(new int[]{Types.INTEGER, Types.VARCHAR, ...}));

        // execute the program
        env.execute("Flink MySQL Example");
    }
}

注意，需要将代码中的 `mydatabase`、`mytable`、`myuser` 和 `mypassword` 替换为你自己的数据库名称、表名称和登录凭据。

此外，还需要根据你的表结构修改 `setQuery` 和 `setSqlTypes` 方法中的参数。在 `setSqlTypes` 中，可以使用 `java.sql.Types` 中的常量指定每个列的 SQL 类型。

最后，运行程序并检查新的 MySQL 数据库是否包含来自源数据库的数据。