flink中数据写入到mysql中

在 Flink 中将数据写入到 MySQL 中，需要使用 Flink 的 JDBC 连接器。步骤如下：

1. 导入 JDBC 驱动

在 Flink 的 pom.xml 文件中加入 MySQL 的 JDBC 驱动依赖，例如：

<dependencies>
    <!-- ... -->
    <dependency>
        <groupId>mysql</groupId>
        <artifactId>mysql-connector-java</artifactId>
        <version>8.0.25</version>
    </dependency>
</dependencies>

2. 编写 MySQL 输出格式化器

Flink 的 JDBC 连接器需要使用一个输出格式化器将数据格式化为 SQL 语句。可以自定义一个实现了 `JDBCOutputFormat` 接口的输出格式化器，例如：

public class MySQLOutputFormat extends JDBCOutputFormat {
    public MySQLOutputFormat(String username, String password, String drivername, String dbURL, String query) {
        super(username, password, drivername, dbURL, query);
    }

    @Override
    protected void prepareStatement(PreparedStatement preparedStatement, Tuple tuple) throws SQLException {
        // 将 tuple 中的数据设置到 PreparedStatement 中
        preparedStatement.setInt(1, tuple.getField(0));
        preparedStatement.setString(2, tuple.getField(1));
        preparedStatement.setDouble(3, tuple.getField(2));
    }
}

3. 将数据写入到 MySQL

在 Flink 的数据流中，使用 `addSink` 方法将数据写入到 MySQL 中，例如：

DataStream<Tuple3<Integer, String, Double>> dataStream = ...;

dataStream.addSink(
    new JDBCOutputFormat(
        "jdbc:mysql://localhost:3306/test",
        "root",
        "password",
        "INSERT INTO table_name (id, name, value) VALUES (?, ?, ?)"
    )
);

其中，`dataStream` 是一个包含了 3 个字段的数据流，分别为整数、字符串和双精度浮点数。`JDBCOutputFormat` 的参数依次为 MySQL 的 JDBC URL、用户名、密码和 SQL 插入语句。在 `prepareStatement` 中，将 tuple 中的字段设置到 PreparedStatement 中，即将数据写入到 MySQL 中。