输出到MySQL

1. 添加依赖

<dependency><groupId>org.apache.flink</groupId><artifactId>flink-connector-jdbc</artifactId><version>3.1.0-1.17</version>
</dependency>
<dependency><groupId>com.mysql</groupId><artifactId>mysql-connector-j</artifactId><version>8.0.32</version>
</dependency>

2. 启动MySQL, 在test库下建表clicks

CREATE TABLE `clicks` (`user` VARCHAR(100) NOT NULL,`url` VARCHAR(100) DEFAULT NULL,`ts` BIGINT DEFAULT NULL
) ENGINE=INNODB DEFAULT CHARSET=utf8

3. 示例代码

public class Flink04_JdbcSink {public static void main(String[] args) {//1.创建运行环境StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();//默认是最大并行度env.setParallelism(1);DataStreamSource<Event> ds = Flink06_EventSource.getEventSource(env);//SinkFunction<Event> sink = JdbcSink.sink("insert into clicks(user, url, ts) values (?,?,?)", new JdbcStatementBuilder<Event>() {@Overridepublic void accept(PreparedStatement preparedStatement, Event event) throws SQLException {//给SQL的占位符赋值preparedStatement.setString(1, event.getUser());preparedStatement.setString(2, event.getUrl());preparedStatement.setLong(3, event.getTs());}},JdbcExecutionOptions.builder().withBatchSize(5).withBatchIntervalMs(10000).withMaxRetries(3).build(),new JdbcConnectionOptions.JdbcConnectionOptionsBuilder().withDriverName("com.mysql.cj.jdbc.Driver").withUsername("root").withPassword("000000").withUrl("jdbc:mysql://hadoop102:3306/flink").build());ds.addSink(sink);try {env.execute();} catch (Exception e) {throw new RuntimeException(e);}}
}

MySQL的幂等性处理

1. 将插入关键字替换为replace，如果主键重复，将除了主键外的所有字段都替换。
2. 使用on duplicate key update 字段名 = values(字段名)语法，如果主键重复，可以选择部分字段进行替换，其余字段保持不变。
3. 示例代码

public class Flink05_JdbcSinkReplace {public static void main(String[] args) {//1.创建运行环境StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();//默认是最大并行度env.setParallelism(1);DataStreamSource<Event> ds = Flink06_EventSource.getEventSource(env);SingleOutputStreamOperator<WordCount> countDs =ds.map(event -> new WordCount(event.getUrl(), 1)).keyBy(WordCount::getWord).sum("count");//SinkFunction<WordCount> sink = JdbcSink.sink(
//                "replace into url_count(url, cnt) values (?,?)""insert into url_count(url, cnt) values(?,?) on duplicate key update cnt = values(cnt)",new JdbcStatementBuilder<WordCount>() {@Overridepublic void accept(PreparedStatement preparedStatement, WordCount wordCount) throws SQLException {//注意：这里的起始下标是1preparedStatement.setString(1, wordCount.getWord());preparedStatement.setInt(2, wordCount.getCount());}},JdbcExecutionOptions.builder().withBatchSize(5).withBatchIntervalMs(10000).withMaxRetries(3).build(),new JdbcConnectionOptions.JdbcConnectionOptionsBuilder().withDriverName("com.mysql.cj.jdbc.Driver").withUsername("root").withPassword("000000").withUrl("jdbc:mysql://hadoop102:3306/flink").build());countDs.addSink(sink);try {env.execute();} catch (Exception e) {throw new RuntimeException(e);}}
}