flink: 通过Sink将数据写入MySQL

在Apache Flink中，要通过Sink将数据写入MySQL，你可以使用JDBC连接器。以下是一个简单的例子，演示如何将数据写入MySQL数据库。

首先，确保你的项目中包含了Flink的JDBC模块依赖。以下是Maven依赖的一个示例：

<dependency>
    <groupId>org.apache.flink</groupId>
    <artifactId>flink-connector-jdbc_2.11</artifactId>
    <version>1.13.0</version> <!--请替换为你使用的Flink版本-->
</dependency>

然后，你可以在Flink程序中添加如下的Sink：

import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.connectors.jdbc.JDBCOutputFormat;
 
public class FlinkJdbcSinkExample {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
 
        DataStream<Tuple2<String, Integer>> inputDataStream = env.fromElements(
            Tuple2.of("user1", 25),
            Tuple2.of("user2", 30)
            // ... 更多的Tuple2<String, Integer>元素
        );
 
        JDBCOutputFormat jdbcOutputFormat = JDBCOutputFormat.buildJDBCOutputFormat()
            .setDrivername("com.mysql.jdbc.Driver")
            .setDBUrl("jdbc:mysql://localhost:3306/database_name")
            .setUsername("username")
            .setPassword("password")
            .setQuery("INSERT INTO table_name (column1, column2) VALUES (?, ?)")
            .setSqlTypes(String.class, Integer.class)
            .build();
 
        inputDataStream.output(jdbcOutputFormat);
 
        env.execute("Flink JDBC Sink Example");
    }
}

在上面的代码中，你需要替换database_name、username、password以及table_name为你的MySQL数据库信息，并且确保column1和column2与你的表结构相匹配。

请注意，这个例子假设你的MySQL表已经存在，并且有正确的列和数据类型。如果需要，你可以在执行Flink程序之前通过SQL语句创建表。