輸出到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);}}
}